From 9eb8bc5109d8f14e6443ed611fdac561ef0dbad9 Mon Sep 17 00:00:00 2001 From: scially Date: Thu, 9 Apr 2026 18:21:06 +0800 Subject: [PATCH] Add QGIS agent harness --- .gitignore | 4 + QGIS/agent-harness/QGIS.md | 127 +++ .../agent-harness/cli_anything/qgis/README.md | 183 +++++ .../cli_anything/qgis/__init__.py | 3 + .../cli_anything/qgis/__main__.py | 7 + .../cli_anything/qgis/core/__init__.py | 1 + .../cli_anything/qgis/core/export.py | 130 +++ .../cli_anything/qgis/core/features.py | 109 +++ .../cli_anything/qgis/core/layers.py | 232 ++++++ .../cli_anything/qgis/core/layouts.py | 191 +++++ .../cli_anything/qgis/core/processing.py | 78 ++ .../cli_anything/qgis/core/project.py | 153 ++++ .../cli_anything/qgis/core/session.py | 142 ++++ .../cli_anything/qgis/qgis_cli.py | 750 ++++++++++++++++++ .../cli_anything/qgis/skills/SKILL.md | 140 ++++ .../cli_anything/qgis/tests/TEST.md | 157 ++++ .../cli_anything/qgis/tests/test_core.py | 259 ++++++ .../cli_anything/qgis/tests/test_full_e2e.py | 562 +++++++++++++ .../cli_anything/qgis/utils/__init__.py | 1 + .../cli_anything/qgis/utils/qgis_backend.py | 247 ++++++ .../cli_anything/qgis/utils/repl_skin.py | 193 +++++ QGIS/agent-harness/pytest.ini | 2 + QGIS/agent-harness/setup.py | 52 ++ 23 files changed, 3723 insertions(+) create mode 100644 QGIS/agent-harness/QGIS.md create mode 100644 QGIS/agent-harness/cli_anything/qgis/README.md create mode 100644 QGIS/agent-harness/cli_anything/qgis/__init__.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/__main__.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/__init__.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/export.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/features.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/layers.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/layouts.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/processing.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/project.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/core/session.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/qgis_cli.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/skills/SKILL.md create mode 100644 QGIS/agent-harness/cli_anything/qgis/tests/TEST.md create mode 100644 QGIS/agent-harness/cli_anything/qgis/tests/test_core.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/utils/__init__.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/utils/qgis_backend.py create mode 100644 QGIS/agent-harness/cli_anything/qgis/utils/repl_skin.py create mode 100644 QGIS/agent-harness/pytest.ini create mode 100644 QGIS/agent-harness/setup.py diff --git a/.gitignore b/.gitignore index e2b01227a..14ea40781 100644 --- a/.gitignore +++ b/.gitignore @@ -92,6 +92,7 @@ !/renderdoc/ !/cloudcompare/ !/openscreen/ +!/QGIS/ !/n8n/ !/obsidian/ @@ -164,6 +165,8 @@ /cloudcompare/.* /openscreen/* /openscreen/.* +/QGIS/* +/QGIS/.* /cloudanalyzer/* /cloudanalyzer/.* /wiremock/* @@ -217,6 +220,7 @@ !/wiremock/ !/wiremock/agent-harness/ !/exa/agent-harness/ +!/QGIS/agent-harness/ !/n8n/agent-harness/ !/obsidian/agent-harness/ !/safari/ diff --git a/QGIS/agent-harness/QGIS.md b/QGIS/agent-harness/QGIS.md new file mode 100644 index 000000000..8d1c588d6 --- /dev/null +++ b/QGIS/agent-harness/QGIS.md @@ -0,0 +1,127 @@ +# QGIS harness architecture notes + +This harness targets the real QGIS runtime already present on the machine and follows the cli-anything harness model: keep authoring state inside a long-lived Python process, and use existing backend CLIs where QGIS already exposes them cleanly. + +## Backend split + +### PyQGIS for stateful authoring + +Use PyQGIS for operations that mutate or inspect in-memory project state: + +- project create/open/save +- project CRS/title updates +- writable vector layer creation +- feature insertion +- layout creation and item authoring +- project/layer/layout summaries for CLI output + +Key APIs and source references: + +- `QgsApplication` + - `QGIS/src/core/qgsapplication.h` + - `QGIS/src/core/qgsapplication.cpp` +- `QgsProject` + - `QGIS/src/core/project/qgsproject.h` + - `QGIS/src/core/project/qgsproject.cpp` +- `QgsPrintLayout` / layout items + - runtime API surfaced via PyQGIS +- `QgsLayoutExporter` + - `QGIS/src/core/layout/qgslayoutexporter.h` + +### `qgis_process --json` for processing and export + +Use `qgis_process --json` for operations that already exist as stable QGIS processing algorithms: + +- algorithm discovery (`list`) +- algorithm help (`help `) +- generic algorithm execution (`run `) +- layout PDF export (`native:printlayouttopdf`) +- layout image export (`native:printlayouttoimage`) + +Key references: + +- process CLI entrypoint + - `QGIS/src/process/main.cpp` + - `QGIS/src/process/qgsprocess.cpp` + - `QGIS/src/process/qgsprocess.h` +- layout export algorithms + - `QGIS/src/analysis/processing/qgsalgorithmlayouttopdf.cpp` + - `QGIS/src/analysis/processing/qgsalgorithmlayouttoimage.cpp` +- exporter tests/examples + - `QGIS/tests/src/python/test_qgslayoutexporter.py` + +## Why this split + +PyQGIS is the right layer for commands that need live project state across multiple commands in a REPL. `qgis_process` is the right layer for processing algorithms because QGIS already ships a supported CLI contract, including JSON output and algorithm metadata. + +This keeps the harness close to QGIS instead of reimplementing backend logic in Python. + +## How the harness was surveyed + +This harness was not generated automatically from `QGIS/src`. The command groups were chosen by surveying QGIS' stable runtime surfaces and then wrapping a narrow, testable subset. + +- `project`, `layer`, `feature`, and `layout` wrap PyQGIS authoring surfaces that need live project state +- `export` and `process` wrap stable `qgis_process --json` surfaces instead of reimplementing algorithms in Python +- `session` is harness-local ergonomics for REPL/history tracking, not a native QGIS feature + +This also explains the current boundaries: + +- the harness intentionally exposes a small layout surface instead of the full desktop layout system +- the harness can run processing algorithms directly against shapefiles and other datasource paths +- the harness does not currently provide a first-class command to add an arbitrary existing shapefile into a project + +## Data model choices + +### Projects + +Projects are saved immediately to `.qgz` or `.qgs` paths. The harness normalizes bare output names to `.qgz` by default. + +### Writable layers + +New vector layers are created as project-side GeoPackage layers instead of ephemeral memory layers. The default datastore is derived from the project path: + +- `.qgz` -> `_data.gpkg` + +This keeps authored data on disk and makes subsequent processing/export commands work against real datasets. + +### Features + +Feature insertion accepts: + +- WKT geometry +- repeatable `--attr key=value` + +Attributes are coerced against the declared QGIS field types so CLI input remains simple while the stored layer schema stays authoritative. + +### Layouts + +The harness only implements a narrow but useful layout surface in v1: + +- create/remove/list layouts +- add map items +- add label items +- derive map extent from current project layers when no explicit extent is given + +## Session model + +The CLI maintains lightweight session state for: + +- current project path +- modified flag reporting +- command history + +The REPL is the default mode when no subcommand is passed. One-shot commands can still use `--project` to bind a command to a saved project path. + +## Output model + +Every command supports a stable JSON shape through `--json`. Human-readable output is intentionally secondary; agent-facing usage should prefer JSON. + +## Testing strategy + +The test suite should cover three layers: + +1. direct module tests against PyQGIS helpers +2. real E2E flows against the actual QGIS runtime +3. subprocess tests against the installed `cli-anything-qgis` executable + +That combination checks both the library layer and the packaging/runtime contract. diff --git a/QGIS/agent-harness/cli_anything/qgis/README.md b/QGIS/agent-harness/cli_anything/qgis/README.md new file mode 100644 index 000000000..dbfbc1f45 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/README.md @@ -0,0 +1,183 @@ +# cli-anything-qgis + +A stateful CLI harness for QGIS that uses the real QGIS runtime. + +## What it does + +- manages `.qgs` / `.qgz` projects with PyQGIS +- creates writable GeoPackage-backed vector layers +- adds and inspects features with WKT geometry +- authors simple print layouts +- exports layouts through `qgis_process --json` +- exposes generic QGIS processing discovery and execution +- supports machine-readable `--json` output for every command +- starts a stateful REPL when run without a subcommand + +## Runtime model + +This harness deliberately splits responsibilities: + +- **PyQGIS** handles project, layer, feature, and layout authoring +- **`qgis_process --json`** handles generic processing and layout export algorithms + +That keeps authoring stateful while still using QGIS' existing processing CLI for backend execution. + +## Prerequisites + +- QGIS installed with `qgis_process` on `PATH` +- PyQGIS importable from the same Python used to run this package +- Python 3.10+ + +Quick checks: + +```bash +qgis_process --version +python3 -c "from qgis.core import QgsApplication; print('pyqgis-ok')" +``` + +## Installation + +```bash +cd /home/wangh68/project/cli_anything_g/QGIS/agent-harness +python3 -m pip install -e . +``` + +If PyQGIS comes from your system packages, a plain virtual environment may not see the `qgis` Python module. In that case, create the environment with system site packages enabled before installing: + +```bash +python3 -m venv --system-site-packages .venv +. .venv/bin/activate +python3 -m pip install -e . +``` + +Verify: + +```bash +which cli-anything-qgis +cli-anything-qgis --help +cli-anything-qgis --json process help native:printlayouttopdf +``` + +## Agent guidance + +Prefer `--json` unless a human-readable summary is specifically more useful. + +## More docs + +- architecture note: [`../../QGIS.md`](../../QGIS.md) +- Chinese walkthrough and real-data tutorial: [`../../TUTORIAL.md`](../../TUTORIAL.md) + +## Usage + +### One-shot commands + +```bash +# Create a project +cli-anything-qgis --json project new -o demo.qgz --title "Demo" --crs EPSG:4326 + +# Create a writable layer in the project's sidecar GeoPackage +cli-anything-qgis --json --project demo.qgz layer create-vector \ + --name places \ + --geometry point \ + --field name:string \ + --field score:int + +# Add features by WKT +cli-anything-qgis --json --project demo.qgz feature add \ + --layer places \ + --wkt "POINT(1 2)" \ + --attr name=HQ \ + --attr score=5 + +# Create a layout and add items +cli-anything-qgis --json --project demo.qgz layout create --name Main +# Auto extent should work even for point-only projects; pass --extent only when you want explicit framing. +cli-anything-qgis --json --project demo.qgz layout add-map --layout Main --x 10 --y 20 --width 180 --height 120 +cli-anything-qgis --json --project demo.qgz layout add-label --layout Main --text "Demo map" --x 10 --y 8 --width 120 --height 10 + +# Export through the real QGIS backend +cli-anything-qgis --json --project demo.qgz export pdf output.pdf --layout Main --overwrite +cli-anything-qgis --json --project demo.qgz export image output.png --layout Main --overwrite + +# Inspect processing algorithms +cli-anything-qgis --json process list +cli-anything-qgis --json process help native:buffer +cli-anything-qgis --json --project demo.qgz process run native:buffer \ + --param INPUT=places \ + --param DISTANCE=10 \ + --param SEGMENTS=8 \ + --param END_CAP_STYLE=0 \ + --param JOIN_STYLE=0 \ + --param MITER_LIMIT=2 \ + --param DISSOLVE=false \ + --param OUTPUT=/tmp/buffer.gpkg +``` + +### REPL + +Run without arguments: + +```bash +cli-anything-qgis +``` + +Example session: + +```text +project new -o demo.qgz --title "Demo" +layer create-vector --name places --geometry point --field name:string +feature add --layer places --wkt "POINT(1 2)" --attr name=HQ +layout create --name Main +layout add-map --layout Main --x 10 --y 20 --width 180 --height 120 +export pdf demo.pdf --layout Main --overwrite +session status +quit +``` + +## Command groups + +### `project` +- `new` — create a new project and save it immediately +- `open` — open an existing project +- `save` — save the active project +- `info` — inspect the current project +- `set-crs` — change project CRS + +### `layer` +- `create-vector` — create a GeoPackage-backed vector layer +- `list` — list project layers +- `info` — inspect one layer +- `remove` — remove a layer from the project + +### `feature` +- `add` — add a feature with WKT geometry and `key=value` attrs +- `list` — inspect features from a layer + +### `layout` +- `create` — create a print layout +- `list` — list layouts +- `info` — inspect one layout +- `remove` — remove a layout +- `add-map` — add a map item +- `add-label` — add a text label item + +### `export` +- `presets` — list supported export modes +- `pdf` — export a layout as PDF +- `image` — export a layout as an image + +### `process` +- `list` — list installed processing algorithms +- `help` — inspect algorithm parameters and outputs +- `run` — execute a processing algorithm with repeatable `--param KEY=VALUE` + +### `session` +- `status` — inspect current session state +- `history` — inspect recent command history + +## Testing + +```bash +python3 -m pytest cli_anything/qgis/tests/test_core.py -v +python3 -m pytest cli_anything/qgis/tests/test_full_e2e.py -v -s +``` diff --git a/QGIS/agent-harness/cli_anything/qgis/__init__.py b/QGIS/agent-harness/cli_anything/qgis/__init__.py new file mode 100644 index 000000000..9bb330bb4 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/__init__.py @@ -0,0 +1,3 @@ +"""cli-anything-qgis package.""" + +__version__ = "1.0.0" diff --git a/QGIS/agent-harness/cli_anything/qgis/__main__.py b/QGIS/agent-harness/cli_anything/qgis/__main__.py new file mode 100644 index 000000000..10513b3b8 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/__main__.py @@ -0,0 +1,7 @@ +"""Run cli-anything-qgis as a module.""" + +from cli_anything.qgis.qgis_cli import main + + +if __name__ == "__main__": + main() diff --git a/QGIS/agent-harness/cli_anything/qgis/core/__init__.py b/QGIS/agent-harness/cli_anything/qgis/core/__init__.py new file mode 100644 index 000000000..81a0804f6 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/__init__.py @@ -0,0 +1 @@ +"""Core domain modules for cli-anything-qgis.""" diff --git a/QGIS/agent-harness/cli_anything/qgis/core/export.py b/QGIS/agent-harness/cli_anything/qgis/core/export.py new file mode 100644 index 000000000..d9663f316 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/export.py @@ -0,0 +1,130 @@ +"""Layout export helpers for cli-anything-qgis.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.core.layouts import get_layout +from cli_anything.qgis.utils import qgis_backend as backend +from cli_anything.qgis.utils.qgis_backend import QgisBackendError + + +def _normalize_output_path(path: str) -> str: + return str(Path(path).expanduser().resolve()) + + +def _bool_param(value: bool) -> str: + return "true" if value else "false" + + +def _prepare_output(path: str, overwrite: bool) -> str: + output_path = _normalize_output_path(path) + target = Path(output_path) + target.parent.mkdir(parents=True, exist_ok=True) + if target.exists(): + if not overwrite: + raise QgisBackendError( + f"Output already exists: {output_path}. Use --overwrite to replace it." + ) + target.unlink() + return output_path + + +def export_presets() -> dict: + """Describe the supported layout export modes.""" + return { + "formats": [ + { + "name": "pdf", + "algorithm": "native:printlayouttopdf", + "description": "Export a named print layout as a PDF file.", + }, + { + "name": "image", + "algorithm": "native:printlayouttoimage", + "description": "Export a named print layout as an image file such as PNG.", + }, + ] + } + + +def export_layout_pdf( + output_path: str, + *, + layout_name: str, + dpi: float | None = None, + force_vector: bool = False, + force_raster: bool = False, + georeference: bool = True, + overwrite: bool = False, +) -> dict: + """Export a print layout to PDF via qgis_process.""" + get_layout(layout_name) + project_info = project_mod.save_if_dirty() + project_path = project_mod.require_saved_project_path() + output = _prepare_output(output_path, overwrite) + + parameters = [ + f"LAYOUT={layout_name}", + f"OUTPUT={output}", + f"FORCE_VECTOR={_bool_param(force_vector)}", + f"FORCE_RASTER={_bool_param(force_raster)}", + f"GEOREFERENCE={_bool_param(georeference)}", + ] + if dpi is not None: + parameters.append(f"DPI={dpi}") + + payload = backend.run_algorithm( + "native:printlayouttopdf", + parameters=parameters, + project_path=project_path, + ) + + return { + "format": "pdf", + "layout": layout_name, + "output": payload.get("results", {}).get("OUTPUT", output), + "file_size": os.path.getsize(output), + "project": project_info, + "results": payload.get("results", {}), + "log": payload.get("log", []), + } + + +def export_layout_image( + output_path: str, + *, + layout_name: str, + dpi: float | None = None, + overwrite: bool = False, +) -> dict: + """Export a print layout to an image via qgis_process.""" + get_layout(layout_name) + project_info = project_mod.save_if_dirty() + project_path = project_mod.require_saved_project_path() + output = _prepare_output(output_path, overwrite) + + parameters = [ + f"LAYOUT={layout_name}", + f"OUTPUT={output}", + ] + if dpi is not None: + parameters.append(f"DPI={dpi}") + + payload = backend.run_algorithm( + "native:printlayouttoimage", + parameters=parameters, + project_path=project_path, + ) + + return { + "format": "image", + "layout": layout_name, + "output": payload.get("results", {}).get("OUTPUT", output), + "file_size": os.path.getsize(output), + "project": project_info, + "results": payload.get("results", {}), + "log": payload.get("log", []), + } diff --git a/QGIS/agent-harness/cli_anything/qgis/core/features.py b/QGIS/agent-harness/cli_anything/qgis/core/features.py new file mode 100644 index 000000000..da7786362 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/features.py @@ -0,0 +1,109 @@ +"""Feature editing helpers for cli-anything-qgis.""" + +from __future__ import annotations + +from cli_anything.qgis.core.layers import get_layer, layer_summary +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, ensure_qgis_app + + +def _ensure_vector_layer(layer) -> None: + from qgis.core import QgsMapLayerType + + if layer.type() != QgsMapLayerType.VectorLayer: + raise QgisBackendError("This command only supports vector layers") + + +def _coerce_value(raw_value: str, field) -> object: + from qgis.PyQt.QtCore import QMetaType + + meta_type = field.type() + value = raw_value.strip() + + if meta_type == QMetaType.Type.Int: + return int(value) + if meta_type == QMetaType.Type.Double: + return float(value) + if meta_type == QMetaType.Type.Bool: + lowered = value.lower() + if lowered in {"true", "1", "yes", "on"}: + return True + if lowered in {"false", "0", "no", "off"}: + return False + raise QgisBackendError(f"Invalid boolean value: {raw_value}") + return value + + +def _attribute_map(feature, layer) -> dict: + result = {} + for field in layer.fields(): + result[field.name()] = feature[field.name()] + return result + + +def _feature_summary(feature, layer) -> dict: + geometry = feature.geometry() + return { + "id": int(feature.id()), + "geometry_wkt": geometry.asWkt() if geometry and not geometry.isNull() else None, + "attributes": _attribute_map(feature, layer), + } + + +def add_feature(layer_identifier: str, wkt: str, attr_specs: list[str]) -> dict: + """Add a feature to a vector layer using WKT and key=value attributes.""" + ensure_qgis_app() + from qgis.core import QgsFeature, QgsGeometry + + layer = get_layer(layer_identifier) + _ensure_vector_layer(layer) + + geometry = QgsGeometry.fromWkt(wkt) + if geometry.isNull(): + raise QgisBackendError(f"Invalid WKT geometry: {wkt}") + + provided_attrs: dict[str, object] = {} + for spec in attr_specs: + key, separator, raw_value = spec.partition("=") + if not separator or not key.strip(): + raise QgisBackendError( + f"Invalid attribute specification: {spec}. Use key=value." + ) + field_index = layer.fields().indexFromName(key.strip()) + if field_index < 0: + raise QgisBackendError(f"Unknown field: {key.strip()}") + field = layer.fields()[field_index] + provided_attrs[key.strip()] = _coerce_value(raw_value, field) + + feature = QgsFeature(layer.fields()) + feature.setGeometry(geometry) + for field in layer.fields(): + feature[field.name()] = provided_attrs.get(field.name()) + + added, features = layer.dataProvider().addFeatures([feature]) + if not added: + raise QgisBackendError(f"Failed to add feature to layer: {layer.name()}") + + layer.updateExtents() + added_feature = features[0] if features else feature + return { + "layer": layer_summary(layer), + "feature": _feature_summary(added_feature, layer), + } + + +def list_features(layer_identifier: str, limit: int = 20) -> dict: + """List features from a vector layer.""" + layer = get_layer(layer_identifier) + _ensure_vector_layer(layer) + + features = [] + for index, feature in enumerate(layer.getFeatures()): + if limit and index >= limit: + break + features.append(_feature_summary(feature, layer)) + + return { + "layer": layer_summary(layer), + "feature_count": int(layer.featureCount()), + "features": features, + } diff --git a/QGIS/agent-harness/cli_anything/qgis/core/layers.py b/QGIS/agent-harness/cli_anything/qgis/core/layers.py new file mode 100644 index 000000000..77a4ff97f --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/layers.py @@ -0,0 +1,232 @@ +"""Layer lifecycle helpers for cli-anything-qgis.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Iterable + +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, ensure_qgis_app + +FIELD_TYPES = { + "int": ("integer", "Int"), + "integer": ("integer", "Int"), + "double": ("double", "Double"), + "float": ("double", "Double"), + "string": ("string", "QString"), + "str": ("string", "QString"), + "bool": ("bool", "Bool"), + "boolean": ("bool", "Bool"), +} + +GEOMETRY_TYPES = { + "point": "Point", + "line": "LineString", + "linestring": "LineString", + "polygon": "Polygon", +} + + +def _field_type_enum(type_name: str): + from qgis.PyQt.QtCore import QMetaType + + normalized = type_name.strip().lower() + if normalized not in FIELD_TYPES: + raise QgisBackendError( + f"Unsupported field type: {type_name}. Use one of: int, double, string, bool" + ) + + enum_name = FIELD_TYPES[normalized][1] + return getattr(QMetaType.Type, enum_name), FIELD_TYPES[normalized][0] + + +def parse_field_specs(field_specs: Iterable[str]) -> list[dict]: + """Parse repeated name:type field specifications.""" + parsed: list[dict] = [] + seen_names: set[str] = set() + + for spec in field_specs: + name, separator, raw_type = spec.partition(":") + if not separator or not name.strip() or not raw_type.strip(): + raise QgisBackendError( + f"Invalid field specification: {spec}. Use name:type, e.g. name:string" + ) + + field_name = name.strip() + if field_name in seen_names: + raise QgisBackendError(f"Duplicate field name: {field_name}") + + field_type, normalized_type = _field_type_enum(raw_type) + parsed.append( + { + "name": field_name, + "meta_type": field_type, + "type": normalized_type, + } + ) + seen_names.add(field_name) + + return parsed + + +def _all_layers(): + return list(project_mod.current_project().mapLayers().values()) + + +def get_layer(identifier: str): + """Resolve a layer by id or exact name.""" + project = project_mod.current_project() + if identifier in project.mapLayers(): + return project.mapLayer(identifier) + + matches = [layer for layer in project.mapLayers().values() if layer.name() == identifier] + if not matches: + raise QgisBackendError(f"Layer not found: {identifier}") + if len(matches) > 1: + raise QgisBackendError( + f"Layer name is ambiguous: {identifier}. Use the layer id instead." + ) + return matches[0] + + +def _layer_type_name(layer) -> str: + from qgis.core import QgsMapLayerType + + if layer.type() == QgsMapLayerType.VectorLayer: + return "vector" + if layer.type() == QgsMapLayerType.RasterLayer: + return "raster" + return "other" + + +def _field_descriptions(layer) -> list[dict]: + descriptions = [] + for field in layer.fields(): + descriptions.append( + { + "name": field.name(), + "type": field.typeName() or str(field.type()), + } + ) + return descriptions + + +def layer_summary(layer) -> dict: + """Return a stable summary for a QGIS layer.""" + from qgis.core import QgsMapLayerType, QgsWkbTypes + + layer_type = _layer_type_name(layer) + summary = { + "id": layer.id(), + "name": layer.name(), + "type": layer_type, + "provider": layer.providerType(), + "source": layer.source(), + "crs": layer.crs().authid() if layer.crs().isValid() else None, + } + + if layer.type() == QgsMapLayerType.VectorLayer: + summary.update( + { + "geometry_type": QgsWkbTypes.displayString(layer.wkbType()), + "feature_count": int(layer.featureCount()), + "fields": _field_descriptions(layer), + } + ) + else: + summary.update( + { + "geometry_type": None, + "feature_count": None, + "fields": [], + } + ) + + return summary + + +def list_layers() -> dict: + """List all layers in the active project.""" + layers = sorted((layer_summary(layer) for layer in _all_layers()), key=lambda item: item["name"]) + return {"count": len(layers), "layers": layers} + + +def layer_info(identifier: str) -> dict: + """Return detailed information for a single layer.""" + return layer_summary(get_layer(identifier)) + + +def create_vector_layer( + name: str, + geometry: str, + crs: str, + field_specs: Iterable[str], +) -> dict: + """Create a GeoPackage-backed vector layer and add it to the current project.""" + ensure_qgis_app() + from qgis.core import ( + QgsCoordinateReferenceSystem, + QgsField, + QgsVectorFileWriter, + QgsVectorLayer, + ) + + if any(layer.name() == name for layer in _all_layers()): + raise QgisBackendError(f"Layer already exists: {name}") + + geometry_key = geometry.strip().lower() + if geometry_key not in GEOMETRY_TYPES: + raise QgisBackendError( + f"Unsupported geometry type: {geometry}. Use point, linestring, or polygon." + ) + + crs_value = QgsCoordinateReferenceSystem(crs) + if not crs_value.isValid(): + raise QgisBackendError(f"Invalid CRS: {crs}") + + fields = parse_field_specs(field_specs) + project = project_mod.current_project() + datastore_path = project_mod.default_datastore_path() + + memory_layer = QgsVectorLayer(f"{GEOMETRY_TYPES[geometry_key]}?crs={crs}", name, "memory") + if not memory_layer.isValid(): + raise QgisBackendError("Failed to create the in-memory source layer") + + provider = memory_layer.dataProvider() + provider.addAttributes([QgsField(field["name"], field["meta_type"]) for field in fields]) + memory_layer.updateFields() + + options = QgsVectorFileWriter.SaveVectorOptions() + options.driverName = "GPKG" + options.layerName = name + if Path(datastore_path).exists(): + options.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteLayer + + error_code, error_message, new_filename, new_layer = QgsVectorFileWriter.writeAsVectorFormatV3( + memory_layer, + datastore_path, + project.transformContext(), + options, + ) + if error_code != QgsVectorFileWriter.NoError: + raise QgisBackendError(error_message or f"Failed to write layer to {datastore_path}") + + stored_layer = QgsVectorLayer( + f"{new_filename or datastore_path}|layername={new_layer or name}", + name, + "ogr", + ) + if not stored_layer.isValid(): + raise QgisBackendError("Failed to reopen the GeoPackage-backed layer") + + project.addMapLayer(stored_layer) + return layer_summary(stored_layer) + + +def remove_layer(identifier: str) -> dict: + """Remove a layer from the active project.""" + project = project_mod.current_project() + layer = get_layer(identifier) + removed = layer_summary(layer) + project.removeMapLayer(layer.id()) + return removed diff --git a/QGIS/agent-harness/cli_anything/qgis/core/layouts.py b/QGIS/agent-harness/cli_anything/qgis/core/layouts.py new file mode 100644 index 000000000..9a021b6c4 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/layouts.py @@ -0,0 +1,191 @@ +"""Layout authoring helpers for cli-anything-qgis.""" + +from __future__ import annotations + +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, ensure_qgis_app + +PAGE_SIZES = {"A4", "A3", "A2", "A1", "A0", "LETTER"} + + +def _layout_manager(): + return project_mod.current_project().layoutManager() + + +def _all_layouts(): + manager = _layout_manager() + if hasattr(manager, "printLayouts"): + return list(manager.printLayouts()) + if hasattr(manager, "layouts"): + return list(manager.layouts()) + return [] + + +def get_layout(name: str): + """Resolve a print layout by exact name.""" + matches = [layout for layout in _all_layouts() if layout.name() == name] + if not matches: + raise QgisBackendError(f"Layout not found: {name}") + if len(matches) > 1: + raise QgisBackendError(f"Layout name is ambiguous: {name}") + return matches[0] + + +def _item_summary(item) -> dict: + rect = item.sceneBoundingRect() + return { + "type": type(item).__name__, + "display_name": item.displayName() if hasattr(item, "displayName") else type(item).__name__, + "x": round(rect.x(), 2), + "y": round(rect.y(), 2), + "width": round(rect.width(), 2), + "height": round(rect.height(), 2), + } + + +def layout_summary(layout) -> dict: + """Return a stable summary for a print layout.""" + items = [_item_summary(item) for item in layout.items()] + return { + "name": layout.name(), + "item_count": len(items), + "items": items, + } + + +def list_layouts() -> dict: + """List print layouts in the active project.""" + layouts = sorted((layout_summary(layout) for layout in _all_layouts()), key=lambda item: item["name"]) + return {"count": len(layouts), "layouts": layouts} + + +def _combined_project_extent(): + from qgis.core import QgsMapLayerType, QgsRectangle + + extent = None + for layer in project_mod.current_project().mapLayers().values(): + if layer.type() != QgsMapLayerType.VectorLayer and layer.type() != QgsMapLayerType.RasterLayer: + continue + layer_extent = layer.extent() + if layer_extent.isNull() or not layer_extent.isFinite(): + continue + if extent is None: + extent = QgsRectangle(layer_extent) + else: + extent.combineExtentWith(layer_extent) + + if extent is None: + raise QgisBackendError( + "Could not determine a map extent. Add at least one layer or pass --extent explicitly." + ) + return extent + + +def _parse_extent(extent: str): + from qgis.core import QgsRectangle + + parts = [part.strip() for part in extent.split(",") if part.strip()] + if len(parts) != 4: + raise QgisBackendError( + f"Invalid extent: {extent}. Use xmin,ymin,xmax,ymax." + ) + xmin, ymin, xmax, ymax = map(float, parts) + return QgsRectangle(xmin, ymin, xmax, ymax) + + +def create_layout( + name: str, + page_size: str = "A4", + orientation: str = "portrait", +) -> dict: + """Create a new print layout.""" + ensure_qgis_app() + from qgis.core import QgsLayoutItemPage, QgsPrintLayout + + if any(layout.name() == name for layout in _all_layouts()): + raise QgisBackendError(f"Layout already exists: {name}") + + normalized_page_size = page_size.strip().upper() + if normalized_page_size not in PAGE_SIZES: + raise QgisBackendError( + f"Unsupported page size: {page_size}. Use one of: {', '.join(sorted(PAGE_SIZES))}" + ) + + normalized_orientation = orientation.strip().lower() + if normalized_orientation not in {"portrait", "landscape"}: + raise QgisBackendError("Orientation must be portrait or landscape") + + layout = QgsPrintLayout(project_mod.current_project()) + layout.initializeDefaults() + layout.setName(name) + page = layout.pageCollection().page(0) + page_orientation = ( + QgsLayoutItemPage.Landscape + if normalized_orientation == "landscape" + else QgsLayoutItemPage.Portrait + ) + if not page.setPageSize(normalized_page_size, page_orientation): + raise QgisBackendError(f"Failed to set page size: {page_size}") + + _layout_manager().addLayout(layout) + return layout_summary(layout) + + +def layout_info(name: str) -> dict: + """Return detailed information for a named layout.""" + return layout_summary(get_layout(name)) + + +def remove_layout(name: str) -> dict: + """Remove a print layout from the project.""" + layout = get_layout(name) + removed = layout_summary(layout) + _layout_manager().removeLayout(layout) + return removed + + +def add_map_item( + layout_name: str, + x: float, + y: float, + width: float, + height: float, + extent: str | None = None, +) -> dict: + """Add a map item to an existing layout.""" + ensure_qgis_app() + from qgis.core import QgsLayoutItemMap, QgsLayoutPoint, QgsLayoutSize, QgsUnitTypes + + layout = get_layout(layout_name) + map_item = QgsLayoutItemMap(layout) + map_item.attemptMove(QgsLayoutPoint(x, y, QgsUnitTypes.LayoutMillimeters)) + map_item.attemptResize(QgsLayoutSize(width, height, QgsUnitTypes.LayoutMillimeters)) + map_item.setExtent(_parse_extent(extent) if extent else _combined_project_extent()) + layout.addLayoutItem(map_item) + return layout_summary(layout) + + +def add_label_item( + layout_name: str, + text: str, + x: float, + y: float, + width: float, + height: float, + font_size: float = 18.0, +) -> dict: + """Add a label item to an existing layout.""" + ensure_qgis_app() + from qgis.PyQt.QtGui import QFont + from qgis.core import QgsLayoutItemLabel, QgsLayoutPoint, QgsLayoutSize, QgsUnitTypes + + layout = get_layout(layout_name) + label = QgsLayoutItemLabel(layout) + label.setText(text) + font = QFont() + font.setPointSizeF(font_size) + label.setFont(font) + label.attemptMove(QgsLayoutPoint(x, y, QgsUnitTypes.LayoutMillimeters)) + label.attemptResize(QgsLayoutSize(width, height, QgsUnitTypes.LayoutMillimeters)) + layout.addLayoutItem(label) + return layout_summary(layout) diff --git a/QGIS/agent-harness/cli_anything/qgis/core/processing.py b/QGIS/agent-harness/cli_anything/qgis/core/processing.py new file mode 100644 index 000000000..817bfb21c --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/processing.py @@ -0,0 +1,78 @@ +"""Processing wrappers around qgis_process.""" + +from __future__ import annotations + +from cli_anything.qgis.utils import qgis_backend as backend +from cli_anything.qgis.utils.qgis_backend import QgisBackendError + + +def parse_param_specs(param_specs: list[str]) -> list[str]: + """Validate repeated KEY=VALUE parameter specifications.""" + parameters: list[str] = [] + for spec in param_specs: + key, separator, value = spec.partition("=") + if not separator or not key.strip(): + raise QgisBackendError( + f"Invalid parameter specification: {spec}. Use KEY=VALUE." + ) + parameters.append(f"{key.strip()}={value}") + return parameters + + +def list_algorithms() -> dict: + """Return a flattened view of installed QGIS processing algorithms.""" + payload = backend.list_algorithms() + algorithms = [] + + for provider_id, provider in sorted(payload.get("providers", {}).items()): + for algorithm_id, details in sorted(provider.get("algorithms", {}).items()): + algorithms.append( + { + "id": algorithm_id, + "name": details.get("name"), + "provider": provider_id, + "group": details.get("group"), + "short_description": details.get("short_description"), + } + ) + + return { + "qgis_version": payload.get("qgis_version"), + "provider_count": len(payload.get("providers", {})), + "algorithm_count": len(algorithms), + "algorithms": algorithms, + } + + +def help_algorithm(algorithm_id: str) -> dict: + """Return structured help for a processing algorithm.""" + payload = backend.help_algorithm(algorithm_id) + return { + "qgis_version": payload.get("qgis_version"), + "provider": payload.get("provider_details", {}), + "algorithm": payload.get("algorithm_details", {}), + "parameters": payload.get("parameters", []), + "outputs": payload.get("outputs", []), + } + + +def run_algorithm( + algorithm_id: str, + *, + param_specs: list[str], + project_path: str | None = None, +) -> dict: + """Run a processing algorithm through qgis_process.""" + payload = backend.run_algorithm( + algorithm_id, + parameters=parse_param_specs(param_specs), + project_path=project_path, + ) + return { + "qgis_version": payload.get("qgis_version"), + "project_path": payload.get("project_path"), + "algorithm": payload.get("algorithm_details", {}), + "inputs": payload.get("inputs", {}), + "results": payload.get("results", {}), + "log": payload.get("log", []), + } diff --git a/QGIS/agent-harness/cli_anything/qgis/core/project.py b/QGIS/agent-harness/cli_anything/qgis/core/project.py new file mode 100644 index 000000000..aec27b6ce --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/project.py @@ -0,0 +1,153 @@ +"""Project lifecycle helpers for cli-anything-qgis.""" + +from __future__ import annotations + +from pathlib import Path + +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, ensure_qgis_app + + +def normalize_project_path(path: str) -> str: + """Normalize project paths and default to .qgz when no extension is given.""" + target = Path(path).expanduser() + if target.suffix.lower() not in {".qgs", ".qgz"}: + target = target.with_suffix(".qgz") + return str(target.resolve()) + + +def current_project(): + """Return the singleton QgsProject instance.""" + ensure_qgis_app() + from qgis.core import QgsProject + + return QgsProject.instance() + + +def current_project_path() -> str: + """Return the current project file path, if any.""" + return current_project().fileName() or "" + + +def require_saved_project_path() -> str: + """Return the active project path or raise if the project is unsaved.""" + project_path = current_project_path() + if not project_path: + raise QgisBackendError( + "No saved project is loaded. Create or open a project first, or pass --project." + ) + return project_path + + +def default_datastore_path(project_path: str | None = None) -> str: + """Return the default GeoPackage path for a saved project.""" + path = Path(project_path or require_saved_project_path()) + return str(path.with_name(f"{path.stem}_data.gpkg")) + + +def _layout_names(project) -> list[str]: + manager = project.layoutManager() + if hasattr(manager, "printLayouts"): + return sorted(layout.name() for layout in manager.printLayouts()) + if hasattr(manager, "layouts"): + return sorted(layout.name() for layout in manager.layouts()) + return [] + + +def project_info() -> dict: + """Return a summary of the active project.""" + project = current_project() + project_path = project.fileName() or "" + layer_names = sorted(layer.name() for layer in project.mapLayers().values()) + layout_names = _layout_names(project) + + return { + "path": project_path or None, + "title": project.title() or None, + "crs": project.crs().authid() if project.crs().isValid() else None, + "modified": bool(project.isDirty()), + "layer_count": len(layer_names), + "layout_count": len(layout_names), + "layer_names": layer_names, + "layout_names": layout_names, + "datastore_path": default_datastore_path(project_path) if project_path else None, + } + + +def create_project(output_path: str, title: str | None = None, crs: str = "EPSG:4326") -> dict: + """Create a new QGIS project and save it immediately.""" + ensure_qgis_app() + from qgis.core import QgsCoordinateReferenceSystem + + project = current_project() + normalized = normalize_project_path(output_path) + target = Path(normalized) + target.parent.mkdir(parents=True, exist_ok=True) + + project.clear() + project.setFileName(normalized) + project.setTitle(title or target.stem) + + crs_value = QgsCoordinateReferenceSystem(crs) + if not crs_value.isValid(): + raise QgisBackendError(f"Invalid CRS: {crs}") + project.setCrs(crs_value) + + if not project.write(): + raise QgisBackendError(f"Failed to create project at {normalized}") + + return project_info() + + +def open_project(project_path: str) -> dict: + """Load an existing QGIS project.""" + project = current_project() + normalized = normalize_project_path(project_path) + target = Path(normalized) + if not target.exists(): + raise QgisBackendError(f"Project does not exist: {normalized}") + + project.clear() + if not project.read(normalized): + raise QgisBackendError(f"Failed to open project: {normalized}") + + return project_info() + + +def save_project(output_path: str | None = None) -> dict: + """Save the active QGIS project.""" + project = current_project() + target_path = normalize_project_path(output_path) if output_path else current_project_path() + if not target_path: + raise QgisBackendError("No project file path is set. Use project save PATH.") + + target = Path(target_path) + target.parent.mkdir(parents=True, exist_ok=True) + project.setFileName(target_path) + + if not project.write(): + raise QgisBackendError(f"Failed to save project: {target_path}") + + return project_info() + + +def save_if_dirty() -> dict: + """Save the active project when it is dirty or missing on disk.""" + project = current_project() + target_path = require_saved_project_path() + if project.isDirty() or not Path(target_path).exists(): + return save_project(target_path) + return project_info() + + +def set_project_crs(crs: str) -> dict: + """Set the active project CRS.""" + ensure_qgis_app() + from qgis.core import QgsCoordinateReferenceSystem + + project = current_project() + crs_value = QgsCoordinateReferenceSystem(crs) + if not crs_value.isValid(): + raise QgisBackendError(f"Invalid CRS: {crs}") + + project.setCrs(crs_value) + return project_info() diff --git a/QGIS/agent-harness/cli_anything/qgis/core/session.py b/QGIS/agent-harness/cli_anything/qgis/core/session.py new file mode 100644 index 000000000..bc5832645 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/core/session.py @@ -0,0 +1,142 @@ +"""Session state and command history for cli-anything-qgis.""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + + +def _locked_save_json(path: str, data: dict, **dump_kwargs) -> None: + """Atomically write JSON with exclusive file locking.""" + try: + handle = open(path, "r+", encoding="utf-8") + except FileNotFoundError: + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + handle = open(path, "w", encoding="utf-8") + + with handle: + locked = False + try: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + locked = True + except (ImportError, OSError): + pass + + try: + handle.seek(0) + handle.truncate() + json.dump(data, handle, **dump_kwargs) + handle.flush() + finally: + if locked: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +@dataclass +class HistoryEntry: + """A single CLI command execution.""" + + command: str + args: dict + timestamp: str = "" + result: dict | None = None + + def __post_init__(self) -> None: + if not self.timestamp: + self.timestamp = datetime.now(timezone.utc).isoformat() + + def to_dict(self) -> dict: + return { + "command": self.command, + "args": self.args, + "timestamp": self.timestamp, + "result": self.result, + } + + @classmethod + def from_dict(cls, value: dict) -> "HistoryEntry": + return cls( + command=value["command"], + args=value.get("args", {}), + timestamp=value.get("timestamp", ""), + result=value.get("result"), + ) + + +class Session: + """Tracks the current project path and command history.""" + + def __init__(self, session_file: str | None = None): + self.current_project_path = "" + self._history: list[HistoryEntry] = [] + self._session_file = session_file + if session_file: + self._load(session_file) + + @property + def history_count(self) -> int: + return len(self._history) + + @property + def active_project_name(self) -> str: + if not self.current_project_path: + return "" + return Path(self.current_project_path).name + + def set_project_path(self, path: str | None) -> None: + normalized = str(path or "") + if normalized == self.current_project_path: + return + self.current_project_path = normalized + self._auto_save() + + def clear_project(self) -> None: + if not self.current_project_path: + return + self.current_project_path = "" + self._auto_save() + + def record(self, command: str, args: dict, result: dict | None = None) -> None: + self._history.append(HistoryEntry(command=command, args=args, result=result)) + self._auto_save() + + def history(self, limit: int = 20) -> list[dict]: + entries = self._history[-limit:] if limit else self._history + return [entry.to_dict() for entry in entries] + + def status(self, *, modified: bool = False) -> dict: + return { + "current_project_path": self.current_project_path or None, + "project_name": self.active_project_name or None, + "modified": modified, + "history_count": self.history_count, + } + + def save(self, path: str) -> None: + data = { + "current_project_path": self.current_project_path, + "history": [entry.to_dict() for entry in self._history], + } + _locked_save_json(path, data, indent=2, sort_keys=True) + + def _auto_save(self) -> None: + if self._session_file: + self.save(self._session_file) + + def _load(self, path: str) -> None: + session_path = Path(path) + if not session_path.exists(): + return + try: + data = json.loads(session_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return + self.current_project_path = str(data.get("current_project_path", "") or "") + self._history = [HistoryEntry.from_dict(entry) for entry in data.get("history", [])] diff --git a/QGIS/agent-harness/cli_anything/qgis/qgis_cli.py b/QGIS/agent-harness/cli_anything/qgis/qgis_cli.py new file mode 100644 index 000000000..046e06ac5 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/qgis_cli.py @@ -0,0 +1,750 @@ +#!/usr/bin/env python3 +"""QGIS CLI — stateful project, layer, layout, export, and processing commands. + +Usage: + # One-shot commands + cli-anything-qgis project new -o demo.qgz --title "Demo" + cli-anything-qgis --project demo.qgz layer create-vector --name places --geometry point --field name:string + cli-anything-qgis --project demo.qgz feature add --layer places --wkt "POINT(1 2)" --attr name=HQ + cli-anything-qgis --project demo.qgz layout create --name Main + cli-anything-qgis --project demo.qgz export pdf out.pdf --layout Main --overwrite + cli-anything-qgis --json process help native:printlayouttopdf + + # Interactive REPL + cli-anything-qgis +""" + +from __future__ import annotations + +import json +import shlex +import sys +from pathlib import Path +from typing import Optional + +import click + +from cli_anything.qgis import __version__ +from cli_anything.qgis.core import export as export_mod +from cli_anything.qgis.core import features as features_mod +from cli_anything.qgis.core import layers as layers_mod +from cli_anything.qgis.core import layouts as layouts_mod +from cli_anything.qgis.core import processing as processing_mod +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.core.session import Session +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, QgisProcessError + +_session: Optional[Session] = None +_json_output = False +_repl_mode = False + + +def get_session() -> Session: + """Return the process-local session object.""" + global _session + if _session is None: + session_dir = Path.home() / ".cli-anything-qgis" + session_dir.mkdir(parents=True, exist_ok=True) + _session = Session(str(session_dir / "session.json")) + return _session + + +def _print_dict(data: dict, indent: int = 0) -> None: + prefix = " " * indent + for key, value in data.items(): + if isinstance(value, dict): + click.echo(f"{prefix}{key}:") + _print_dict(value, indent + 1) + elif isinstance(value, list): + click.echo(f"{prefix}{key}:") + _print_list(value, indent + 1) + else: + click.echo(f"{prefix}{key}: {value}") + + +def _print_list(items: list, indent: int = 0) -> None: + prefix = " " * indent + for index, item in enumerate(items): + if isinstance(item, dict): + click.echo(f"{prefix}[{index}]") + _print_dict(item, indent + 1) + else: + click.echo(f"{prefix}- {item}") + + +def output(data, message: str = "") -> None: + """Emit data in either JSON or human-readable form.""" + if _json_output: + click.echo(json.dumps(data, indent=2, default=str)) + return + + if message: + click.echo(message) + if isinstance(data, dict): + _print_dict(data) + elif isinstance(data, list): + _print_list(data) + else: + click.echo(str(data)) + + +def _error_payload(exc: Exception) -> dict: + payload = { + "error": str(exc), + "type": exc.__class__.__name__, + } + if isinstance(exc, QgisProcessError): + payload["returncode"] = exc.returncode + if exc.stderr: + payload["stderr"] = exc.stderr + if exc.stdout: + payload["stdout"] = exc.stdout + if exc.payload: + payload["payload"] = exc.payload + return payload + + +def handle_error(func): + """Normalize domain/backend errors for CLI and REPL use.""" + + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except (QgisBackendError, QgisProcessError, ValueError) as exc: + payload = _error_payload(exc) + if _json_output: + click.echo(json.dumps(payload, indent=2, default=str)) + else: + click.echo(f"Error: {exc}", err=True) + if not _repl_mode: + raise SystemExit(1) + return None + + wrapper.__name__ = func.__name__ + wrapper.__doc__ = func.__doc__ + return wrapper + + +def _requested_project_path() -> str | None: + ctx = click.get_current_context(silent=True) + if ctx is None: + return None + root = ctx.find_root() + obj = root.obj or {} + return obj.get("project_path") + + +def _sync_session_project_path() -> None: + session = get_session() + current_path = project_mod.current_project_path() + if current_path: + session.set_project_path(current_path) + else: + session.clear_project() + + +def _current_project_modified() -> bool: + try: + return bool(project_mod.current_project().isDirty()) + except Exception: + return False + + +def _load_requested_project(required: bool = False) -> str | None: + requested = _requested_project_path() + if requested: + normalized = project_mod.normalize_project_path(requested) + if project_mod.current_project_path() != normalized: + project_mod.open_project(normalized) + _sync_session_project_path() + return normalized + + if required and not project_mod.current_project_path(): + raise QgisBackendError( + "No project is loaded. Open one with project open or pass --project." + ) + + return project_mod.current_project_path() or None + + +def _active_project_path(required: bool = False) -> str | None: + requested = _requested_project_path() + if requested: + return project_mod.normalize_project_path(requested) + + current = project_mod.current_project_path() + if current: + return current + + if required: + raise QgisBackendError( + "No project is loaded. Open one with project open or pass --project." + ) + return None + + +def _auto_save_if_one_shot() -> None: + if _repl_mode: + return + if project_mod.current_project_path(): + project_mod.save_project() + _sync_session_project_path() + + +def _record(command: str, args: dict, result=None) -> None: + summary = None + if isinstance(result, dict): + summary_keys = { + "path", + "output", + "format", + "title", + "layer_count", + "layout_count", + "feature_count", + "count", + "name", + } + summary = {key: value for key, value in result.items() if key in summary_keys} + if not summary and "layer" in result and isinstance(result["layer"], dict): + summary = {"layer": result["layer"].get("name")} + get_session().record(command, args, summary) + + +@click.group(invoke_without_command=True) +@click.option("--json", "use_json", is_flag=True, help="Output as JSON") +@click.option( + "--project", + "project_path", + type=click.Path(path_type=Path), + default=None, + help="Open this project for the current command.", +) +@click.pass_context +def cli(ctx, use_json, project_path): + """QGIS CLI for project authoring, layout export, and processing.""" + global _json_output + _json_output = use_json + get_session() + ctx.obj = {"project_path": str(project_path) if project_path else None} + + if ctx.invoked_subcommand is None and "--help" not in sys.argv and "-h" not in sys.argv: + ctx.invoke(repl) + + +@cli.group() +def project(): + """Project management commands.""" + + +@project.command("new") +@click.option("-o", "--output", "output_path", required=True, type=click.Path(path_type=Path)) +@click.option("--title", default=None, help="Project title") +@click.option("--crs", default="EPSG:4326", help="Project CRS, e.g. EPSG:4326") +@handle_error +def project_new(output_path: Path, title: str | None, crs: str): + """Create a new saved QGIS project.""" + data = project_mod.create_project(str(output_path), title=title, crs=crs) + _sync_session_project_path() + _record("project new", {"output": str(output_path), "title": title, "crs": crs}, data) + output(data, f"Created project: {data['path']}") + + +@project.command("open") +@click.argument("project_path", type=click.Path(path_type=Path)) +@handle_error +def project_open(project_path: Path): + """Open an existing QGIS project.""" + data = project_mod.open_project(str(project_path)) + _sync_session_project_path() + _record("project open", {"project_path": str(project_path)}, data) + output(data, f"Opened project: {data['path']}") + + +@project.command("save") +@click.argument("output_path", required=False, type=click.Path(path_type=Path)) +@handle_error +def project_save(output_path: Path | None): + """Save the current QGIS project.""" + if output_path is None: + _load_requested_project(required=bool(_requested_project_path())) + data = project_mod.save_project(str(output_path) if output_path else None) + _sync_session_project_path() + _record("project save", {"output": str(output_path) if output_path else None}, data) + output(data, f"Saved project: {data['path']}") + + +@project.command("info") +@handle_error +def project_info(): + """Show information about the current QGIS project.""" + if _requested_project_path(): + _load_requested_project(required=True) + data = project_mod.project_info() + _record("project info", {}, data) + output(data) + + +@project.command("set-crs") +@click.argument("crs") +@handle_error +def project_set_crs(crs: str): + """Set the active project's CRS.""" + _load_requested_project(required=True) + data = project_mod.set_project_crs(crs) + _auto_save_if_one_shot() + _record("project set-crs", {"crs": crs}, data) + output(data, f"Updated project CRS to {crs}") + + +@cli.group() +def layer(): + """Layer management commands.""" + + +@layer.command("create-vector") +@click.option("--name", required=True, help="Layer name") +@click.option( + "--geometry", + required=True, + type=click.Choice(["point", "linestring", "polygon"], case_sensitive=False), + help="Geometry type", +) +@click.option("--crs", default=None, help="Layer CRS, e.g. EPSG:4326") +@click.option("--field", "field_specs", multiple=True, help="Field spec as name:type") +@handle_error +def layer_create_vector(name: str, geometry: str, crs: str | None, field_specs: tuple[str, ...]): + """Create a GeoPackage-backed vector layer in the current project.""" + _load_requested_project(required=True) + effective_crs = crs or project_mod.project_info().get("crs") or "EPSG:4326" + data = layers_mod.create_vector_layer(name, geometry, effective_crs, field_specs) + _auto_save_if_one_shot() + _record( + "layer create-vector", + { + "name": name, + "geometry": geometry, + "crs": effective_crs, + "fields": list(field_specs), + }, + data, + ) + output(data, f"Created layer: {data['name']}") + + +@layer.command("list") +@handle_error +def layer_list(): + """List layers in the current project.""" + _load_requested_project(required=True) + data = layers_mod.list_layers() + _record("layer list", {}, data) + output(data) + + +@layer.command("info") +@click.argument("identifier") +@handle_error +def layer_info(identifier: str): + """Show detailed information for a layer.""" + _load_requested_project(required=True) + data = layers_mod.layer_info(identifier) + _record("layer info", {"identifier": identifier}, data) + output(data) + + +@layer.command("remove") +@click.argument("identifier") +@handle_error +def layer_remove(identifier: str): + """Remove a layer from the current project.""" + _load_requested_project(required=True) + data = layers_mod.remove_layer(identifier) + _auto_save_if_one_shot() + _record("layer remove", {"identifier": identifier}, data) + output(data, f"Removed layer: {data['name']}") + + +@cli.group() +def feature(): + """Feature editing commands.""" + + +@feature.command("add") +@click.option("--layer", "layer_identifier", required=True, help="Target layer id or name") +@click.option("--wkt", required=True, help="Geometry in WKT format") +@click.option("--attr", "attr_specs", multiple=True, help="Feature attribute as key=value") +@handle_error +def feature_add(layer_identifier: str, wkt: str, attr_specs: tuple[str, ...]): + """Add a feature to a vector layer using WKT geometry.""" + _load_requested_project(required=True) + data = features_mod.add_feature(layer_identifier, wkt, list(attr_specs)) + _auto_save_if_one_shot() + _record( + "feature add", + {"layer": layer_identifier, "wkt": wkt, "attrs": list(attr_specs)}, + data, + ) + output(data, f"Added feature to {data['layer']['name']}") + + +@feature.command("list") +@click.option("--layer", "layer_identifier", required=True, help="Target layer id or name") +@click.option("--limit", default=20, show_default=True, type=int, help="Maximum features to show") +@handle_error +def feature_list(layer_identifier: str, limit: int): + """List features from a vector layer.""" + _load_requested_project(required=True) + data = features_mod.list_features(layer_identifier, limit=limit) + _record("feature list", {"layer": layer_identifier, "limit": limit}, data) + output(data) + + +@cli.group() +def layout(): + """Print layout authoring commands.""" + + +@layout.command("create") +@click.option("--name", required=True, help="Layout name") +@click.option("--page-size", default="A4", show_default=True, help="Page size") +@click.option( + "--orientation", + default="portrait", + show_default=True, + type=click.Choice(["portrait", "landscape"], case_sensitive=False), + help="Page orientation", +) +@handle_error +def layout_create(name: str, page_size: str, orientation: str): + """Create a print layout.""" + _load_requested_project(required=True) + data = layouts_mod.create_layout(name, page_size=page_size, orientation=orientation) + _auto_save_if_one_shot() + _record( + "layout create", + {"name": name, "page_size": page_size, "orientation": orientation}, + data, + ) + output(data, f"Created layout: {name}") + + +@layout.command("list") +@handle_error +def layout_list(): + """List print layouts in the current project.""" + _load_requested_project(required=True) + data = layouts_mod.list_layouts() + _record("layout list", {}, data) + output(data) + + +@layout.command("info") +@click.argument("name") +@handle_error +def layout_info(name: str): + """Show detailed information for a print layout.""" + _load_requested_project(required=True) + data = layouts_mod.layout_info(name) + _record("layout info", {"name": name}, data) + output(data) + + +@layout.command("remove") +@click.argument("name") +@handle_error +def layout_remove(name: str): + """Remove a print layout from the project.""" + _load_requested_project(required=True) + data = layouts_mod.remove_layout(name) + _auto_save_if_one_shot() + _record("layout remove", {"name": name}, data) + output(data, f"Removed layout: {name}") + + +@layout.command("add-map") +@click.option("--layout", "layout_name", required=True, help="Layout name") +@click.option("--x", type=float, required=True, help="Left position in millimeters") +@click.option("--y", type=float, required=True, help="Top position in millimeters") +@click.option("--width", type=float, required=True, help="Item width in millimeters") +@click.option("--height", type=float, required=True, help="Item height in millimeters") +@click.option("--extent", default=None, help="Map extent as xmin,ymin,xmax,ymax") +@handle_error +def layout_add_map(layout_name: str, x: float, y: float, width: float, height: float, extent: str | None): + """Add a map item to a print layout.""" + _load_requested_project(required=True) + data = layouts_mod.add_map_item(layout_name, x, y, width, height, extent=extent) + _auto_save_if_one_shot() + _record( + "layout add-map", + { + "layout": layout_name, + "x": x, + "y": y, + "width": width, + "height": height, + "extent": extent, + }, + data, + ) + output(data, f"Added map item to layout: {layout_name}") + + +@layout.command("add-label") +@click.option("--layout", "layout_name", required=True, help="Layout name") +@click.option("--text", required=True, help="Label text") +@click.option("--x", type=float, required=True, help="Left position in millimeters") +@click.option("--y", type=float, required=True, help="Top position in millimeters") +@click.option("--width", type=float, required=True, help="Item width in millimeters") +@click.option("--height", type=float, required=True, help="Item height in millimeters") +@click.option("--font-size", default=18.0, show_default=True, type=float, help="Font size") +@handle_error +def layout_add_label( + layout_name: str, + text: str, + x: float, + y: float, + width: float, + height: float, + font_size: float, +): + """Add a label item to a print layout.""" + _load_requested_project(required=True) + data = layouts_mod.add_label_item(layout_name, text, x, y, width, height, font_size=font_size) + _auto_save_if_one_shot() + _record( + "layout add-label", + { + "layout": layout_name, + "text": text, + "x": x, + "y": y, + "width": width, + "height": height, + "font_size": font_size, + }, + data, + ) + output(data, f"Added label item to layout: {layout_name}") + + +@cli.group() +def export(): + """Layout export commands.""" + + +@export.command("presets") +@handle_error +def export_presets(): + """List supported export formats and their backend algorithms.""" + data = export_mod.export_presets() + _record("export presets", {}, data) + output(data) + + +@export.command("pdf") +@click.argument("output_path", type=click.Path(path_type=Path)) +@click.option("--layout", "layout_name", required=True, help="Layout name") +@click.option("--dpi", default=None, type=float, help="Override layout DPI") +@click.option("--force-vector", is_flag=True, help="Always export vector output") +@click.option("--force-raster", is_flag=True, help="Always rasterize the PDF") +@click.option("--georeference/--no-georeference", default=True, help="Append georeference metadata") +@click.option("--overwrite", is_flag=True, help="Overwrite existing output") +@handle_error +def export_pdf( + output_path: Path, + layout_name: str, + dpi: float | None, + force_vector: bool, + force_raster: bool, + georeference: bool, + overwrite: bool, +): + """Export a named print layout as PDF.""" + _load_requested_project(required=True) + data = export_mod.export_layout_pdf( + str(output_path), + layout_name=layout_name, + dpi=dpi, + force_vector=force_vector, + force_raster=force_raster, + georeference=georeference, + overwrite=overwrite, + ) + _record( + "export pdf", + { + "output": str(output_path), + "layout": layout_name, + "dpi": dpi, + "force_vector": force_vector, + "force_raster": force_raster, + "georeference": georeference, + "overwrite": overwrite, + }, + data, + ) + output(data, f"Exported PDF: {data['output']}") + + +@export.command("image") +@click.argument("output_path", type=click.Path(path_type=Path)) +@click.option("--layout", "layout_name", required=True, help="Layout name") +@click.option("--dpi", default=None, type=float, help="Override layout DPI") +@click.option("--overwrite", is_flag=True, help="Overwrite existing output") +@handle_error +def export_image(output_path: Path, layout_name: str, dpi: float | None, overwrite: bool): + """Export a named print layout as an image file.""" + _load_requested_project(required=True) + data = export_mod.export_layout_image( + str(output_path), + layout_name=layout_name, + dpi=dpi, + overwrite=overwrite, + ) + _record( + "export image", + { + "output": str(output_path), + "layout": layout_name, + "dpi": dpi, + "overwrite": overwrite, + }, + data, + ) + output(data, f"Exported image: {data['output']}") + + +@cli.group() +def process(): + """Generic qgis_process discovery and execution commands.""" + + +@process.command("list") +@handle_error +def process_list(): + """List installed QGIS processing algorithms.""" + data = processing_mod.list_algorithms() + _record("process list", {}, {"count": data.get("algorithm_count")}) + output(data) + + +@process.command("help") +@click.argument("algorithm_id") +@handle_error +def process_help(algorithm_id: str): + """Show parameter and output details for a processing algorithm.""" + data = processing_mod.help_algorithm(algorithm_id) + _record("process help", {"algorithm_id": algorithm_id}, data) + output(data) + + +@process.command("run") +@click.argument("algorithm_id") +@click.option("--param", "param_specs", multiple=True, help="Algorithm parameter as KEY=VALUE") +@handle_error +def process_run(algorithm_id: str, param_specs: tuple[str, ...]): + """Run a QGIS processing algorithm through qgis_process.""" + data = processing_mod.run_algorithm( + algorithm_id, + param_specs=list(param_specs), + project_path=_active_project_path(required=False), + ) + _record("process run", {"algorithm_id": algorithm_id, "params": list(param_specs)}, data) + output(data) + + +@cli.group() +def session(): + """Session state and history commands.""" + + +@session.command("status") +@handle_error +def session_status(): + """Show the current session status.""" + _sync_session_project_path() + data = get_session().status(modified=_current_project_modified()) + _record("session status", {}, data) + output(data) + + +@session.command("history") +@click.option("--limit", default=20, show_default=True, type=int, help="Maximum history entries to show") +@handle_error +def session_history(limit: int): + """Show recent command history.""" + data = {"history": get_session().history(limit=limit)} + _record("session history", {"limit": limit}, {"count": len(data['history'])}) + output(data) + + +@cli.command() +@handle_error +def repl(): + """Start the interactive REPL.""" + from cli_anything.qgis.utils.repl_skin import ReplSkin + + global _repl_mode + _repl_mode = True + + skin = ReplSkin("qgis", version=__version__) + skin.print_banner() + prompt_session = skin.create_prompt_session() + + command_help = { + "project": "new|open|save|info|set-crs", + "layer": "create-vector|list|info|remove", + "feature": "add|list", + "layout": "create|list|info|remove|add-map|add-label", + "export": "presets|pdf|image", + "process": "list|help|run", + "session": "status|history", + "help": "Show this help", + "quit": "Exit REPL", + } + + try: + while True: + _sync_session_project_path() + session_state = get_session() + line = skin.get_input( + prompt_session, + project_name=session_state.active_project_name, + modified=_current_project_modified(), + ) + if not line: + continue + lowered = line.lower() + if lowered in {"quit", "exit", "q"}: + skin.print_goodbye() + break + if lowered == "help": + skin.help(command_help) + continue + + try: + args = shlex.split(line) + except ValueError as exc: + skin.error(str(exc)) + continue + + try: + cli.main(args=args, standalone_mode=False) + except SystemExit: + pass + except click.ClickException as exc: + skin.error(str(exc)) + except Exception as exc: # pragma: no cover - last-resort REPL guard + skin.error(str(exc)) + finally: + _repl_mode = False + + +def main() -> None: + """CLI entry point for setuptools.""" + cli() + + +if __name__ == "__main__": + main() diff --git a/QGIS/agent-harness/cli_anything/qgis/skills/SKILL.md b/QGIS/agent-harness/cli_anything/qgis/skills/SKILL.md new file mode 100644 index 000000000..8df56e8bf --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/skills/SKILL.md @@ -0,0 +1,140 @@ +--- +name: cli-anything-qgis +description: Stateful QGIS CLI for projects, writable layers, features, layouts, exports, and qgis_process operations using the real QGIS runtime. +--- + +# cli-anything-qgis + +Use this skill when you need to inspect or modify QGIS projects from the terminal through the real QGIS runtime. + +## Requirements + +- QGIS installed with `qgis_process` on `PATH` +- PyQGIS importable from the Python environment running the CLI +- Python 3.10+ + +## Agent guidance + +- Prefer `--json` for all machine-driven use. +- For one-shot commands, pass `--project ` when operating on an existing project. +- Running `cli-anything-qgis` with no subcommand starts a stateful REPL. +- Layout export is backed by real QGIS processing algorithms: + - `native:printlayouttopdf` + - `native:printlayouttoimage` + +## What this CLI covers + +- create, open, save, and inspect `.qgs` / `.qgz` projects +- create writable GeoPackage-backed vector layers +- add features via WKT geometry and typed `key=value` attributes +- create print layouts and add map/label items +- export layouts to PDF or image +- inspect and run generic `qgis_process` algorithms +- inspect session status and history + +## Command groups + +### `project` +- `new -o/--output [--title] [--crs]` +- `open PATH` +- `save [PATH]` +- `info` +- `set-crs CRS` + +### `layer` +- `create-vector --name --geometry --crs [--field name:type ...]` +- `list` +- `info LAYER` +- `remove LAYER` + +### `feature` +- `add --layer LAYER --wkt WKT [--attr key=value ...]` +- `list --layer LAYER [--limit N]` + +### `layout` +- `create --name [--page-size] [--orientation]` +- `list` +- `info LAYOUT` +- `remove LAYOUT` +- `add-map --layout LAYOUT --x --y --width --height [--extent xmin,ymin,xmax,ymax]` +- `add-label --layout LAYOUT --text TEXT --x --y --width --height [--font-size N]` + +### `export` +- `presets` +- `pdf OUTPUT --layout LAYOUT [--dpi] [--force-vector] [--force-raster] [--georeference/--no-georeference] [--overwrite]` +- `image OUTPUT --layout LAYOUT [--dpi] [--overwrite]` + +### `process` +- `list` +- `help ALGORITHM_ID` +- `run ALGORITHM_ID [--param KEY=VALUE ...]` + +### `session` +- `status` +- `history [--limit N]` + +## Examples + +### Create a project and add a writable layer + +```bash +cli-anything-qgis --json project new -o demo.qgz --title "Demo" --crs EPSG:4326 +cli-anything-qgis --json --project demo.qgz layer create-vector \ + --name places \ + --geometry point \ + --field name:string \ + --field score:int +``` + +### Add features and inspect them + +```bash +cli-anything-qgis --json --project demo.qgz feature add \ + --layer places \ + --wkt "POINT(1 2)" \ + --attr name=HQ \ + --attr score=5 + +cli-anything-qgis --json --project demo.qgz feature list --layer places --limit 10 +``` + +### Create and export a layout + +```bash +cli-anything-qgis --json --project demo.qgz layout create --name Main +cli-anything-qgis --json --project demo.qgz layout add-map --layout Main --x 10 --y 20 --width 180 --height 120 +cli-anything-qgis --json --project demo.qgz layout add-label --layout Main --text "Demo map" --x 10 --y 8 --width 100 --height 10 +cli-anything-qgis --json --project demo.qgz export pdf output.pdf --layout Main --overwrite +``` + +### Inspect or run processing algorithms + +```bash +cli-anything-qgis --json process help native:buffer +cli-anything-qgis --json --project demo.qgz process run native:buffer \ + --param INPUT=/tmp/demo_data.gpkg|layername=places \ + --param DISTANCE=1 \ + --param SEGMENTS=8 \ + --param END_CAP_STYLE=0 \ + --param JOIN_STYLE=0 \ + --param MITER_LIMIT=2 \ + --param DISSOLVE=false \ + --param OUTPUT=/tmp/buffer.geojson +``` + +### REPL + +```bash +cli-anything-qgis +``` + +Example interactive flow: + +```text +project new -o demo.qgz --title "Demo" +layer create-vector --name places --geometry point --field name:string +feature add --layer places --wkt "POINT(1 2)" --attr name=HQ +layout create --name Main +session status +quit +``` diff --git a/QGIS/agent-harness/cli_anything/qgis/tests/TEST.md b/QGIS/agent-harness/cli_anything/qgis/tests/TEST.md new file mode 100644 index 000000000..a34283aa2 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/tests/TEST.md @@ -0,0 +1,157 @@ +# cli-anything-qgis test plan + +## Scope + +This test plan covers the production-style QGIS harness under `cli_anything/qgis/`. + +The suite is split into: + +- `test_core.py` for direct module tests against PyQGIS helpers and backend wrappers +- `test_full_e2e.py` for real workflow coverage and installed-command subprocess coverage + +## Planned unit and module coverage + +### Backend helpers +- `find_qgis_process()` returns a usable executable path or raises a clear error +- `project_path_argument()` normalizes project paths +- `run_process_json()` normalizes JSON and failure cases + +### Project helpers +- create a new project +- open an existing project +- save a project to a path +- change project CRS +- derive the default datastore path +- summarize project metadata + +### Layer helpers +- parse field specs +- create GeoPackage-backed vector layers with fields +- list and inspect layers +- remove layers + +### Feature helpers +- add features using WKT and typed attributes +- list features with limits +- validate bad attr specifications and bad booleans + +### Layout helpers +- create layouts with page size/orientation +- list and inspect layouts +- add map items +- add label items +- remove layouts + +### Session helpers +- record command history +- save/load session history +- report session status + +## Planned real end-to-end workflows + +### Workflow 1 — scratch project to PDF +1. create a new project +2. create a writable vector layer +3. add features +4. create a layout +5. add a map item and label +6. export PDF +7. verify file exists, has non-zero size, and starts with `%PDF-` + +### Workflow 2 — scratch project to PNG +1. create a new project +2. create content and layout +3. export image +4. verify file exists, PNG signature is valid, and dimensions are positive + +### Workflow 3 — processing passthrough +1. create a vector layer and add features +2. run `native:buffer` through the harness +3. verify output dataset exists and opens as a valid vector layer +4. verify buffered feature count is positive + +## Planned subprocess coverage + +Subprocess tests must target the installed executable via `_resolve_cli("cli-anything-qgis")`. + +Planned checks: + +- `cli-anything-qgis --help` +- `cli-anything-qgis --json process help native:printlayouttopdf` +- `cli-anything-qgis --json project new -o ...` +- full installed-command PDF workflow +- full installed-command PNG workflow + +## Planned execution commands + +```bash +CLI_ANYTHING_FORCE_INSTALLED=1 python3 -m pytest cli_anything/qgis/tests/test_core.py -v +CLI_ANYTHING_FORCE_INSTALLED=1 python3 -m pytest cli_anything/qgis/tests/test_full_e2e.py -v -s +CLI_ANYTHING_FORCE_INSTALLED=1 python3 -m pytest cli_anything/qgis/tests -v -s --tb=no +``` + +## Results + +### Coverage note + +- Direct core tests covered project, layer, feature, layout, session, and backend error-normalization helpers, including a point-only layout auto-extent regression. +- Real E2E coverage exercised PDF export, PNG export, and `native:buffer` processing against the installed QGIS runtime. +- Subprocess coverage verified the installed `cli-anything-qgis` entrypoint, JSON help output, project creation, full PDF/PNG workflows, and the point-only `layout add-map` regression path. +- One known warning remains: `QgsLayoutItemLabel.setFont()` is deprecated in this QGIS build, but exports and tests pass. + +### `pytest -v -s --tb=no` output + +```text +============================= test session starts ============================== +platform linux -- Python 3.12.3, pytest-9.0.2, pluggy-1.6.0 -- /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/python +cachedir: .pytest_cache +rootdir: /home/wangh68/project/cli_anything_g/QGIS/agent-harness +configfile: pytest.ini +plugins: cov-7.1.0, anyio-4.12.1 +collecting ... collected 22 items + +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_project_create_save_open_and_info PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_default_datastore_path PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_parse_field_and_param_specs PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_layer_create_list_info_and_remove PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_feature_add_and_list PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_feature_add_rejects_invalid_boolean PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_layout_create_add_items_and_remove PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_layout_add_map_accepts_point_only_project PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_export_presets_describe_supported_formats PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_session_save_load_and_status PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_project_path_argument_normalizes PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_find_qgis_process_missing_raises_clear_error PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_core.py::test_run_process_json_normalizes_backend_failure PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_scratch_project_to_pdf PDF artifact: /tmp/pytest-of-wangh68/pytest-10/test_scratch_project_to_pdf0/workflow.pdf +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_scratch_project_to_png PNG artifact: /tmp/pytest-of-wangh68/pytest-10/test_scratch_project_to_png0/workflow.png +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_processing_passthrough_buffer PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_help [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_process_help_json [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_project_new_json [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_full_pdf_workflow [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +Subprocess PDF artifact: /tmp/pytest-of-wangh68/pytest-10/test_full_pdf_workflow0/subprocess.pdf +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_full_png_workflow [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +Subprocess PNG artifact: /tmp/pytest-of-wangh68/pytest-10/test_full_png_workflow0/subprocess.png +PASSED +QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py::TestCLISubprocess::test_point_only_project_add_map_without_extent [_resolve_cli] Using sibling command: /home/wangh68/project/cli_anything_g/QGIS/agent-harness/.venv/bin/cli-anything-qgis +Subprocess point-only PDF artifact: /tmp/pytest-of-wangh68/pytest-10/test_point_only_project_add_ma0/subprocess_point.pdf +PASSED + +=============================== warnings summary =============================== +cli_anything/qgis/tests/test_core.py::test_layout_create_add_items_and_remove +cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_scratch_project_to_pdf +cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_scratch_project_to_png +cli_anything/qgis/tests/test_full_e2e.py::TestRealCLIWorkflows::test_processing_passthrough_buffer + /home/wangh68/project/cli_anything_g/QGIS/agent-harness/cli_anything/qgis/core/layouts.py:187: DeprecationWarning: QgsLayoutItemLabel.setFont() is deprecated + label.setFont(font) + +-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html +======================= 22 passed, 4 warnings in 18.95s ======================== +``` diff --git a/QGIS/agent-harness/cli_anything/qgis/tests/test_core.py b/QGIS/agent-harness/cli_anything/qgis/tests/test_core.py new file mode 100644 index 000000000..5ff497007 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/tests/test_core.py @@ -0,0 +1,259 @@ +"""Core tests for cli-anything-qgis using PyQGIS and focused backend mocks.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path +from unittest import mock + +import pytest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +_PACKAGE_NAMESPACE_ROOT = Path(__file__).resolve().parents[2] +if str(_PACKAGE_NAMESPACE_ROOT) in sys.path: + sys.path.remove(str(_PACKAGE_NAMESPACE_ROOT)) + +from cli_anything.qgis.core import export as export_mod +from cli_anything.qgis.core import features as features_mod +from cli_anything.qgis.core import layers as layers_mod +from cli_anything.qgis.core import layouts as layouts_mod +from cli_anything.qgis.core import processing as processing_mod +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.core.session import Session +from cli_anything.qgis.utils import qgis_backend as backend +from cli_anything.qgis.utils.qgis_backend import QgisBackendError, QgisProcessError + + +@pytest.fixture(autouse=True) +def clean_qgis_project(): + backend.ensure_qgis_app() + project = project_mod.current_project() + project.clear() + project.setFileName("") + yield + project.clear() + project.setFileName("") + + +def _create_polygon_project(tmp_path: Path, name: str = "demo") -> Path: + project_path = tmp_path / f"{name}.qgz" + project_mod.create_project(str(project_path), title=name.title(), crs="EPSG:4326") + layers_mod.create_vector_layer( + "areas", + "polygon", + "EPSG:4326", + ["name:string", "active:bool", "count:int"], + ) + features_mod.add_feature( + "areas", + "POLYGON((0 0,0 5,5 5,5 0,0 0))", + ["name=ZoneA", "active=true", "count=3"], + ) + project_mod.save_project() + return project_path + + +def test_project_create_save_open_and_info(tmp_path: Path): + project_path = tmp_path / "sample.qgz" + + created = project_mod.create_project(str(project_path), title="Sample", crs="EPSG:4326") + assert created["path"] == str(project_path.resolve()) + assert created["title"] == "Sample" + assert created["crs"] == "EPSG:4326" + assert created["layer_count"] == 0 + assert created["layout_count"] == 0 + assert created["datastore_path"] == str((tmp_path / "sample_data.gpkg").resolve()) + + updated = project_mod.set_project_crs("EPSG:3857") + assert updated["crs"] == "EPSG:3857" + + saved = project_mod.save_project() + assert Path(saved["path"]).exists() + + renamed_path = tmp_path / "renamed.qgz" + renamed = project_mod.save_project(str(renamed_path)) + assert renamed["path"] == str(renamed_path.resolve()) + assert Path(renamed["path"]).exists() + + reopened = project_mod.open_project(str(renamed_path)) + assert reopened["path"] == str(renamed_path.resolve()) + assert reopened["crs"] == "EPSG:3857" + assert reopened["title"] == "Sample" + + +def test_default_datastore_path(tmp_path: Path): + project_path = tmp_path / "city.qgz" + expected = tmp_path / "city_data.gpkg" + assert project_mod.default_datastore_path(str(project_path)) == str(expected) + + +def test_parse_field_and_param_specs(): + fields = layers_mod.parse_field_specs(["name:string", "score:int", "active:bool"]) + assert [field["name"] for field in fields] == ["name", "score", "active"] + assert [field["type"] for field in fields] == ["string", "integer", "bool"] + + params = processing_mod.parse_param_specs(["INPUT=areas", "DISTANCE=10"]) + assert params == ["INPUT=areas", "DISTANCE=10"] + + with pytest.raises(QgisBackendError): + layers_mod.parse_field_specs(["name:string", "name:int"]) + + with pytest.raises(QgisBackendError): + processing_mod.parse_param_specs(["NOT_A_PARAM"]) + + +def test_layer_create_list_info_and_remove(tmp_path: Path): + project_mod.create_project(str(tmp_path / "layers.qgz"), title="Layers", crs="EPSG:4326") + + created = layers_mod.create_vector_layer( + "places", + "point", + "EPSG:4326", + ["name:string", "score:int"], + ) + assert created["name"] == "places" + assert created["provider"] == "ogr" + assert created["type"] == "vector" + assert {"name", "score"}.issubset({field["name"] for field in created["fields"]}) + + listing = layers_mod.list_layers() + assert listing["count"] == 1 + assert listing["layers"][0]["name"] == "places" + + info = layers_mod.layer_info("places") + assert info["id"] == created["id"] + assert info["source"].endswith("layers_data.gpkg|layername=places") + + removed = layers_mod.remove_layer("places") + assert removed["name"] == "places" + assert layers_mod.list_layers()["count"] == 0 + + +def test_feature_add_and_list(tmp_path: Path): + project_mod.create_project(str(tmp_path / "features.qgz"), title="Features", crs="EPSG:4326") + layers_mod.create_vector_layer( + "points", + "point", + "EPSG:4326", + ["name:string", "count:int", "rating:double", "active:bool"], + ) + + added = features_mod.add_feature( + "points", + "POINT(1 2)", + ["name=HQ", "count=7", "rating=2.5", "active=true"], + ) + attrs = added["feature"]["attributes"] + assert attrs["name"] == "HQ" + assert attrs["count"] == 7 + assert float(attrs["rating"]) == pytest.approx(2.5) + assert bool(attrs["active"]) is True + + listing = features_mod.list_features("points", limit=1) + assert listing["feature_count"] == 1 + assert len(listing["features"]) == 1 + assert listing["features"][0]["geometry_wkt"].startswith("Point") + + +def test_feature_add_rejects_invalid_boolean(tmp_path: Path): + project_mod.create_project(str(tmp_path / "invalid_bool.qgz"), title="InvalidBool", crs="EPSG:4326") + layers_mod.create_vector_layer("points", "point", "EPSG:4326", ["active:bool"]) + + with pytest.raises(QgisBackendError): + features_mod.add_feature("points", "POINT(0 0)", ["active=maybe"]) + + +def test_layout_create_add_items_and_remove(tmp_path: Path): + _create_polygon_project(tmp_path, name="layout_demo") + + created = layouts_mod.create_layout("Main", page_size="A4", orientation="portrait") + assert created["name"] == "Main" + + with_map = layouts_mod.add_map_item("Main", 10, 20, 180, 120) + assert any(item["type"] == "QgsLayoutItemMap" for item in with_map["items"]) + + with_label = layouts_mod.add_label_item("Main", "Demo map", 10, 8, 80, 10, font_size=16) + assert any(item["type"] == "QgsLayoutItemLabel" for item in with_label["items"]) + + listing = layouts_mod.list_layouts() + assert listing["count"] == 1 + assert listing["layouts"][0]["name"] == "Main" + + removed = layouts_mod.remove_layout("Main") + assert removed["name"] == "Main" + assert layouts_mod.list_layouts()["count"] == 0 + + +def test_layout_add_map_accepts_point_only_project(tmp_path: Path): + project_mod.create_project(str(tmp_path / "point_layout.qgz"), title="PointLayout", crs="EPSG:4326") + layers_mod.create_vector_layer("points", "point", "EPSG:4326", ["name:string"]) + features_mod.add_feature("points", "POINT(1 2)", ["name=HQ"]) + + layouts_mod.create_layout("Main", page_size="A4", orientation="portrait") + with_map = layouts_mod.add_map_item("Main", 10, 20, 180, 120) + + assert any(item["type"] == "QgsLayoutItemMap" for item in with_map["items"]) + + +def test_export_presets_describe_supported_formats(): + presets = export_mod.export_presets() + formats = {item["name"]: item["algorithm"] for item in presets["formats"]} + assert formats == { + "pdf": "native:printlayouttopdf", + "image": "native:printlayouttoimage", + } + + +def test_session_save_load_and_status(tmp_path: Path): + session_path = tmp_path / "session.json" + session = Session(str(session_path)) + session.set_project_path("/tmp/demo.qgz") + session.record("project info", {"project": "/tmp/demo.qgz"}, {"path": "/tmp/demo.qgz"}) + + reloaded = Session(str(session_path)) + assert reloaded.current_project_path == "/tmp/demo.qgz" + assert reloaded.history_count == 1 + assert reloaded.history(limit=1)[0]["command"] == "project info" + + status = reloaded.status(modified=True) + assert status == { + "current_project_path": "/tmp/demo.qgz", + "project_name": "demo.qgz", + "modified": True, + "history_count": 1, + } + + +def test_project_path_argument_normalizes(tmp_path: Path): + project_path = tmp_path / "demo.qgz" + expected = f"--PROJECT_PATH={project_path.resolve()}" + assert backend.project_path_argument(project_path) == expected + assert backend.project_path_argument(None) is None + + +def test_find_qgis_process_missing_raises_clear_error(): + with mock.patch("cli_anything.qgis.utils.qgis_backend.shutil.which", return_value=None): + with pytest.raises(QgisBackendError, match="qgis_process is not installed"): + backend.find_qgis_process() + + +def test_run_process_json_normalizes_backend_failure(): + payload = {"log": [{"message": "buffer failed"}]} + completed = subprocess.CompletedProcess( + args=["/usr/bin/qgis_process", "--json", "run", "native:buffer"], + returncode=1, + stdout=json.dumps(payload), + stderr="", + ) + + with mock.patch("cli_anything.qgis.utils.qgis_backend.find_qgis_process", return_value="/usr/bin/qgis_process"): + with mock.patch("cli_anything.qgis.utils.qgis_backend.subprocess.run", return_value=completed): + with pytest.raises(QgisProcessError, match="buffer failed") as exc_info: + backend.run_process_json(["run", "native:buffer"]) + + assert exc_info.value.returncode == 1 + assert exc_info.value.payload == payload diff --git a/QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py b/QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py new file mode 100644 index 000000000..10d6ce0c1 --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/tests/test_full_e2e.py @@ -0,0 +1,562 @@ +"""End-to-end and subprocess tests for cli-anything-qgis.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path + +import pytest +from click.testing import CliRunner + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +_PACKAGE_NAMESPACE_ROOT = Path(__file__).resolve().parents[2] +if str(_PACKAGE_NAMESPACE_ROOT) in sys.path: + sys.path.remove(str(_PACKAGE_NAMESPACE_ROOT)) + +from cli_anything.qgis import qgis_cli +from cli_anything.qgis.core import project as project_mod +from cli_anything.qgis.qgis_cli import cli +from cli_anything.qgis.utils import qgis_backend as backend + + +PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" + + +def _resolve_cli(name: str) -> list[str]: + """Resolve the CLI entry-point for subprocess tests. + + Prefers an installed command on PATH and falls back to ``python -m`` + unless ``CLI_ANYTHING_FORCE_INSTALLED=1`` is set. + """ + force = os.environ.get("CLI_ANYTHING_FORCE_INSTALLED", "").strip() == "1" + sibling = Path(sys.executable).parent / name + if sibling.exists(): + print(f"[_resolve_cli] Using sibling command: {sibling}") + return [str(sibling)] + path = shutil.which(name) + if path: + print(f"[_resolve_cli] Using installed command: {path}") + return [path] + if force: + raise RuntimeError(f"{name} not found in PATH. Install with: python3 -m pip install -e .") + module = "cli_anything.qgis.qgis_cli" + print(f"[_resolve_cli] Falling back to: {sys.executable} -m {module}") + return [sys.executable, "-m", module] + + +@pytest.fixture(autouse=True) +def clean_qgis_state(monkeypatch, tmp_path): + home_dir = tmp_path / "home" + home_dir.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("QT_QPA_PLATFORM", os.environ.get("QT_QPA_PLATFORM", "offscreen")) + + backend.ensure_qgis_app() + project = project_mod.current_project() + project.clear() + project.setFileName("") + qgis_cli._session = None + qgis_cli._json_output = False + qgis_cli._repl_mode = False + + yield + + project.clear() + project.setFileName("") + qgis_cli._session = None + qgis_cli._json_output = False + qgis_cli._repl_mode = False + + +@pytest.fixture +def runner() -> CliRunner: + return CliRunner() + + +def _parse_json_output(raw: str) -> dict: + return json.loads(raw) + + +def _invoke_json(runner: CliRunner, args: list[str]) -> dict: + result = runner.invoke(cli, ["--json", *args]) + assert result.exit_code == 0, result.output + return _parse_json_output(result.output) + + +def _subprocess_json(command: list[str], args: list[str], env: dict[str, str]) -> dict: + completed = subprocess.run( + [*command, "--json", *args], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert completed.returncode == 0, completed.stderr or completed.stdout + return json.loads(completed.stdout) + + +def _build_cli_project(runner: CliRunner, tmp_path: Path, stem: str) -> dict[str, str]: + project_path = tmp_path / f"{stem}.qgz" + + _invoke_json(runner, ["project", "new", "-o", str(project_path), "--title", stem]) + layer = _invoke_json( + runner, + [ + "--project", + str(project_path), + "layer", + "create-vector", + "--name", + "areas", + "--geometry", + "polygon", + "--field", + "name:string", + "--field", + "score:int", + ], + ) + _invoke_json( + runner, + [ + "--project", + str(project_path), + "feature", + "add", + "--layer", + "areas", + "--wkt", + "POLYGON((0 0,0 5,5 5,5 0,0 0))", + "--attr", + "name=ZoneA", + "--attr", + "score=5", + ], + ) + _invoke_json(runner, ["--project", str(project_path), "layout", "create", "--name", "Main"]) + _invoke_json( + runner, + [ + "--project", + str(project_path), + "layout", + "add-map", + "--layout", + "Main", + "--x", + "10", + "--y", + "20", + "--width", + "180", + "--height", + "120", + ], + ) + _invoke_json( + runner, + [ + "--project", + str(project_path), + "layout", + "add-label", + "--layout", + "Main", + "--text", + "Demo map", + "--x", + "10", + "--y", + "8", + "--width", + "100", + "--height", + "10", + ], + ) + + return { + "project_path": str(project_path), + "layer_source": layer["source"], + } + + +def _build_subprocess_project(command: list[str], tmp_path: Path, stem: str, env: dict[str, str]) -> dict[str, str]: + project_path = tmp_path / f"{stem}.qgz" + + _subprocess_json(command, ["project", "new", "-o", str(project_path), "--title", stem], env) + layer = _subprocess_json( + command, + [ + "--project", + str(project_path), + "layer", + "create-vector", + "--name", + "areas", + "--geometry", + "polygon", + "--field", + "name:string", + "--field", + "score:int", + ], + env, + ) + _subprocess_json( + command, + [ + "--project", + str(project_path), + "feature", + "add", + "--layer", + "areas", + "--wkt", + "POLYGON((0 0,0 5,5 5,5 0,0 0))", + "--attr", + "name=ZoneA", + "--attr", + "score=5", + ], + env, + ) + _subprocess_json(command, ["--project", str(project_path), "layout", "create", "--name", "Main"], env) + _subprocess_json( + command, + [ + "--project", + str(project_path), + "layout", + "add-map", + "--layout", + "Main", + "--x", + "10", + "--y", + "20", + "--width", + "180", + "--height", + "120", + ], + env, + ) + _subprocess_json( + command, + [ + "--project", + str(project_path), + "layout", + "add-label", + "--layout", + "Main", + "--text", + "Demo map", + "--x", + "10", + "--y", + "8", + "--width", + "100", + "--height", + "10", + ], + env, + ) + + return { + "project_path": str(project_path), + "layer_source": layer["source"], + } + + +def _build_subprocess_point_project( + command: list[str], tmp_path: Path, stem: str, env: dict[str, str] +) -> dict[str, str]: + project_path = tmp_path / f"{stem}.qgz" + + _subprocess_json(command, ["project", "new", "-o", str(project_path), "--title", stem], env) + layer = _subprocess_json( + command, + [ + "--project", + str(project_path), + "layer", + "create-vector", + "--name", + "places", + "--geometry", + "point", + "--field", + "name:string", + "--field", + "score:int", + ], + env, + ) + _subprocess_json( + command, + [ + "--project", + str(project_path), + "feature", + "add", + "--layer", + "places", + "--wkt", + "POINT(116.397 39.907)", + "--attr", + "name=Beijing", + "--attr", + "score=5", + ], + env, + ) + _subprocess_json(command, ["--project", str(project_path), "layout", "create", "--name", "Main"], env) + + return { + "project_path": str(project_path), + "layer_source": layer["source"], + } + + +class TestRealCLIWorkflows: + def test_scratch_project_to_pdf(self, runner: CliRunner, tmp_path: Path): + build = _build_cli_project(runner, tmp_path, "pdf_workflow") + pdf_path = tmp_path / "workflow.pdf" + + exported = _invoke_json( + runner, + [ + "--project", + build["project_path"], + "export", + "pdf", + str(pdf_path), + "--layout", + "Main", + "--overwrite", + ], + ) + + output_path = Path(exported["output"]) + assert output_path.exists() + assert output_path.stat().st_size > 0 + assert output_path.read_bytes()[:5] == b"%PDF-" + print(f"PDF artifact: {output_path}") + + def test_scratch_project_to_png(self, runner: CliRunner, tmp_path: Path): + from qgis.PyQt.QtGui import QImage + + build = _build_cli_project(runner, tmp_path, "png_workflow") + png_path = tmp_path / "workflow.png" + + exported = _invoke_json( + runner, + [ + "--project", + build["project_path"], + "export", + "image", + str(png_path), + "--layout", + "Main", + "--overwrite", + ], + ) + + output_path = Path(exported["output"]) + assert output_path.exists() + assert output_path.read_bytes()[:8] == PNG_SIGNATURE + + image = QImage(str(output_path)) + assert not image.isNull() + assert image.width() > 0 + assert image.height() > 0 + print(f"PNG artifact: {output_path}") + + def test_processing_passthrough_buffer(self, runner: CliRunner, tmp_path: Path): + from qgis.core import QgsVectorLayer + + build = _build_cli_project(runner, tmp_path, "buffer_workflow") + output_path = tmp_path / "buffer.geojson" + + data = _invoke_json( + runner, + [ + "--project", + build["project_path"], + "process", + "run", + "native:buffer", + "--param", + f"INPUT={build['layer_source']}", + "--param", + "DISTANCE=1", + "--param", + "SEGMENTS=8", + "--param", + "END_CAP_STYLE=0", + "--param", + "JOIN_STYLE=0", + "--param", + "MITER_LIMIT=2", + "--param", + "DISSOLVE=false", + "--param", + f"OUTPUT={output_path}", + ], + ) + + result_path = Path(data["results"]["OUTPUT"]) + assert result_path.exists() + + backend.ensure_qgis_app() + layer = QgsVectorLayer(str(result_path), "buffer", "ogr") + assert layer.isValid() + assert int(layer.featureCount()) > 0 + + +class TestCLISubprocess: + def test_help(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + completed = subprocess.run([*command, "--help"], capture_output=True, text=True, check=False, env=env) + assert completed.returncode == 0 + assert "QGIS CLI" in completed.stdout + + def test_process_help_json(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + payload = _subprocess_json(command, ["process", "help", "native:printlayouttopdf"], env) + assert payload["algorithm"]["id"] == "native:printlayouttopdf" + + def test_project_new_json(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + project_path = tmp_path / "subprocess.qgz" + payload = _subprocess_json(command, ["project", "new", "-o", str(project_path), "--title", "Subprocess"], env) + assert payload["path"] == str(project_path.resolve()) + assert Path(payload["path"]).exists() + + def test_full_pdf_workflow(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + build = _build_subprocess_project(command, tmp_path, "subprocess_pdf", env) + pdf_path = tmp_path / "subprocess.pdf" + exported = _subprocess_json( + command, + [ + "--project", + build["project_path"], + "export", + "pdf", + str(pdf_path), + "--layout", + "Main", + "--overwrite", + ], + env, + ) + + output_path = Path(exported["output"]) + assert output_path.exists() + assert output_path.stat().st_size > 0 + assert output_path.read_bytes()[:5] == b"%PDF-" + print(f"Subprocess PDF artifact: {output_path}") + + def test_full_png_workflow(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + build = _build_subprocess_project(command, tmp_path, "subprocess_png", env) + png_path = tmp_path / "subprocess.png" + exported = _subprocess_json( + command, + [ + "--project", + build["project_path"], + "export", + "image", + str(png_path), + "--layout", + "Main", + "--overwrite", + ], + env, + ) + + output_path = Path(exported["output"]) + assert output_path.exists() + assert output_path.read_bytes()[:8] == PNG_SIGNATURE + print(f"Subprocess PNG artifact: {output_path}") + + def test_point_only_project_add_map_without_extent(self, tmp_path: Path): + command = _resolve_cli("cli-anything-qgis") + env = os.environ.copy() + env["HOME"] = str(tmp_path / "home") + env.setdefault("QT_QPA_PLATFORM", "offscreen") + + build = _build_subprocess_point_project(command, tmp_path, "subprocess_point", env) + add_map = _subprocess_json( + command, + [ + "--project", + build["project_path"], + "layout", + "add-map", + "--layout", + "Main", + "--x", + "10", + "--y", + "20", + "--width", + "180", + "--height", + "120", + ], + env, + ) + assert any(item["type"] == "QgsLayoutItemMap" for item in add_map["items"]) + + pdf_path = tmp_path / "subprocess_point.pdf" + exported = _subprocess_json( + command, + [ + "--project", + build["project_path"], + "export", + "pdf", + str(pdf_path), + "--layout", + "Main", + "--overwrite", + ], + env, + ) + + output_path = Path(exported["output"]) + assert output_path.exists() + assert output_path.read_bytes()[:5] == b"%PDF-" + print(f"Subprocess point-only PDF artifact: {output_path}") diff --git a/QGIS/agent-harness/cli_anything/qgis/utils/__init__.py b/QGIS/agent-harness/cli_anything/qgis/utils/__init__.py new file mode 100644 index 000000000..42d4651ad --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/utils/__init__.py @@ -0,0 +1 @@ +"""Utility helpers for cli-anything-qgis.""" diff --git a/QGIS/agent-harness/cli_anything/qgis/utils/qgis_backend.py b/QGIS/agent-harness/cli_anything/qgis/utils/qgis_backend.py new file mode 100644 index 000000000..c44ed677f --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/utils/qgis_backend.py @@ -0,0 +1,247 @@ +"""Low-level backend helpers for PyQGIS and qgis_process.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any + +_QGIS_APP = None + + +class QgisBackendError(RuntimeError): + """Base error raised by the QGIS backend wrapper.""" + + +class QgisProcessError(QgisBackendError): + """Raised when qgis_process fails.""" + + def __init__( + self, + message: str, + *, + command: list[str], + returncode: int, + stdout: str, + stderr: str, + payload: dict[str, Any] | None = None, + ) -> None: + super().__init__(message) + self.command = command + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + self.payload = payload + + +def _normalize_path(path: str | os.PathLike[str]) -> str: + return str(Path(path).expanduser().resolve()) + + +def _detect_qgis_prefix() -> str: + prefix = os.environ.get("QGIS_PREFIX_PATH") + if prefix: + return prefix + + for binary in ("qgis_process", "qgis"): + found = shutil.which(binary) + if found: + return str(Path(found).resolve().parent.parent) + + return "/usr" + + +def find_qgis_process() -> str: + """Return the qgis_process executable path or raise a clear error.""" + path = shutil.which("qgis_process") + if path: + return path + raise QgisBackendError( + "qgis_process is not installed or not available in PATH. Install QGIS and ensure " + "the qgis_process command is available. Example: apt install qgis" + ) + + +def _is_shadow_qgis_module(module: Any) -> bool: + shadow_package = Path(__file__).resolve().parents[1] + + module_file = getattr(module, "__file__", "") or "" + if module_file: + try: + if Path(module_file).resolve().is_relative_to(shadow_package): + return True + except OSError: + pass + + module_paths = getattr(module, "__path__", None) or [] + for item in module_paths: + try: + if Path(item).resolve().is_relative_to(shadow_package): + return True + except OSError: + continue + + return False + + + +def _import_qgs_application(): + shadow_root = str(Path(__file__).resolve().parents[2]) + + for name, module in list(sys.modules.items()): + if (name == "qgis" or name.startswith("qgis.")) and _is_shadow_qgis_module(module): + sys.modules.pop(name, None) + + removed_shadow_root = False + if shadow_root in sys.path: + sys.path.remove(shadow_root) + removed_shadow_root = True + + try: + from qgis.core import QgsApplication + except ImportError as exc: + raise QgisBackendError( + "PyQGIS is not importable in this Python environment. Install QGIS with Python " + "bindings and use a Python interpreter that can import qgis.core." + ) from exc + finally: + if removed_shadow_root: + sys.path.insert(0, shadow_root) + + return QgsApplication + + + +def ensure_qgis_app(): + """Initialize PyQGIS once for this Python process.""" + global _QGIS_APP + + if _QGIS_APP is not None: + return _QGIS_APP + + QgsApplication = _import_qgs_application() + QgsApplication.setPrefixPath(_detect_qgis_prefix(), True) + _QGIS_APP = QgsApplication([], False) + _QGIS_APP.initQgis() + return _QGIS_APP + + +def project_path_argument(project_path: str | None) -> str | None: + """Normalize project paths for qgis_process invocations.""" + if not project_path: + return None + return f"--PROJECT_PATH={_normalize_path(project_path)}" + + +def _extract_payload_message(payload: dict[str, Any] | None) -> str | None: + if not payload: + return None + + log_entries = payload.get("log") + if isinstance(log_entries, list): + parts: list[str] = [] + for entry in log_entries: + if isinstance(entry, dict): + message = entry.get("message") or entry.get("text") + if message: + parts.append(str(message)) + elif entry: + parts.append(str(entry)) + if parts: + return " | ".join(parts[-3:]) + + results = payload.get("results") + if isinstance(results, dict) and "error" in results: + return str(results["error"]) + + return None + + +def run_process_json( + arguments: list[str], + *, + project_path: str | None = None, + parameters: list[str] | None = None, +) -> dict[str, Any]: + """Run qgis_process in JSON mode and return parsed output.""" + command = [find_qgis_process(), "--json", *arguments] + project_arg = project_path_argument(project_path) + if project_arg: + command.append(project_arg) + if parameters: + command.append("--") + command.extend(parameters) + + completed = subprocess.run( + command, + capture_output=True, + text=True, + check=False, + ) + + stdout = completed.stdout.strip() + stderr = completed.stderr.strip() + payload: dict[str, Any] | None = None + + if stdout: + try: + parsed = json.loads(stdout) + if isinstance(parsed, dict): + payload = parsed + else: + payload = {"data": parsed} + except json.JSONDecodeError: + payload = None + + if completed.returncode != 0: + message = _extract_payload_message(payload) or stderr or stdout or ( + f"qgis_process exited with status {completed.returncode}" + ) + raise QgisProcessError( + message, + command=command, + returncode=completed.returncode, + stdout=stdout, + stderr=stderr, + payload=payload, + ) + + if payload is None: + raise QgisProcessError( + "qgis_process returned non-JSON output in --json mode", + command=command, + returncode=completed.returncode, + stdout=stdout, + stderr=stderr, + payload=None, + ) + + return payload + + +def list_algorithms() -> dict[str, Any]: + """Return the raw qgis_process list payload.""" + return run_process_json(["list"]) + + +def help_algorithm(algorithm_id: str) -> dict[str, Any]: + """Return the raw qgis_process help payload for an algorithm.""" + return run_process_json(["help", algorithm_id]) + + +def run_algorithm( + algorithm_id: str, + *, + parameters: list[str] | None = None, + project_path: str | None = None, +) -> dict[str, Any]: + """Run a qgis_process algorithm and return the raw JSON payload.""" + return run_process_json( + ["run", algorithm_id], + project_path=project_path, + parameters=parameters, + ) diff --git a/QGIS/agent-harness/cli_anything/qgis/utils/repl_skin.py b/QGIS/agent-harness/cli_anything/qgis/utils/repl_skin.py new file mode 100644 index 000000000..9ab867d8a --- /dev/null +++ b/QGIS/agent-harness/cli_anything/qgis/utils/repl_skin.py @@ -0,0 +1,193 @@ +"""Shared cli-anything REPL skin for cli-anything-qgis.""" + +from __future__ import annotations + +import os +import sys + +_RESET = "\033[0m" +_BOLD = "\033[1m" +_DARK_GRAY = "\033[38;5;240m" +_GRAY = "\033[38;5;245m" +_LIGHT_GRAY = "\033[38;5;250m" +_CYAN = "\033[38;5;80m" +_GREEN = "\033[38;5;78m" +_YELLOW = "\033[38;5;220m" +_RED = "\033[38;5;196m" + +_ACCENT_COLORS = { + "qgis": "\033[38;5;34m", +} +_DEFAULT_ACCENT = "\033[38;5;75m" + +_ICON_SMALL = "▸" +_H_LINE = "─" +_V_LINE = "│" +_TL = "╭" +_TR = "╮" +_BL = "╰" +_BR = "╯" + + +def _strip_ansi(text: str) -> str: + import re + + return re.sub(r"\033\[[^m]*m", "", text) + + +def _visible_len(text: str) -> int: + return len(_strip_ansi(text)) + + +class ReplSkin: + """Small terminal UI wrapper for the QGIS REPL.""" + + def __init__(self, software: str, version: str = "1.0.0", history_file: str | None = None): + self.software = software.lower().replace("-", "_") + self.display_name = software.replace("_", " ").title() + self.version = version + self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT) + + if history_file is None: + from pathlib import Path + + history_dir = Path.home() / f".cli-anything-{self.software}" + history_dir.mkdir(parents=True, exist_ok=True) + self.history_file = str(history_dir / "history") + else: + self.history_file = history_file + + self._color = self._detect_color_support() + + def _detect_color_support(self) -> bool: + if os.environ.get("NO_COLOR") or os.environ.get("CLI_ANYTHING_NO_COLOR"): + return False + if not hasattr(sys.stdout, "isatty"): + return False + return sys.stdout.isatty() + + def _c(self, code: str, text: str) -> str: + if not self._color: + return text + return f"{code}{text}{_RESET}" + + def print_banner(self) -> None: + inner = 54 + + def box_line(content: str) -> str: + padding = inner - _visible_len(content) + return f"{self._c(_DARK_GRAY, _V_LINE)}{content}{' ' * max(0, padding)}{self._c(_DARK_GRAY, _V_LINE)}" + + top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}") + bottom = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}") + icon = self._c(_CYAN + _BOLD, "◆") + brand = self._c(_CYAN + _BOLD, "cli-anything") + name = self._c(self.accent + _BOLD, self.display_name) + dot = self._c(_DARK_GRAY, "·") + + print(top) + print(box_line(f" {icon} {brand} {dot} {name}")) + print(box_line(f" {self._c(_DARK_GRAY, f'v{self.version}')}")) + print(box_line("")) + print(box_line(f" {self._c(_DARK_GRAY, 'Type help for commands, quit to exit')}")) + print(bottom) + print() + + def prompt(self, project_name: str = "", modified: bool = False, context: str = "") -> str: + parts = [] + parts.append(self._c(_CYAN, "◆ ") if self._color else "> ") + parts.append(self._c(self.accent + _BOLD, self.software)) + if project_name or context: + current = context or project_name + suffix = "*" if modified else "" + parts.append(f" {self._c(_DARK_GRAY, '[')}") + parts.append(self._c(_LIGHT_GRAY, f"{current}{suffix}")) + parts.append(self._c(_DARK_GRAY, ']')) + parts.append(self._c(_GRAY, " ❯ ")) + return "".join(parts) + + def prompt_tokens(self, project_name: str = "", modified: bool = False, context: str = ""): + tokens = [("class:icon", "◆ "), ("class:software", self.software)] + if project_name or context: + current = context or project_name + suffix = "*" if modified else "" + tokens.extend( + [ + ("class:bracket", " ["), + ("class:context", f"{current}{suffix}"), + ("class:bracket", "]"), + ] + ) + tokens.append(("class:arrow", " ❯ ")) + return tokens + + def get_prompt_style(self): + try: + from prompt_toolkit.styles import Style + except ImportError: + return None + + accent = _ANSI_256_TO_HEX.get(self.accent, "#5fafff") + return Style.from_dict( + { + "icon": "#5fd7d7 bold", + "software": f"{accent} bold", + "bracket": "#585858", + "context": "#bcbcbc", + "arrow": "#808080", + "completion-menu.completion": "bg:#303030 #bcbcbc", + "completion-menu.completion.current": f"bg:{accent} #000000", + "completion-menu.meta.completion": "bg:#303030 #808080", + "completion-menu.meta.completion.current": f"bg:{accent} #000000", + "auto-suggest": "#585858", + } + ) + + def error(self, message: str) -> None: + print(f" {self._c(_RED + _BOLD, '✗')} {self._c(_RED, message)}", file=sys.stderr) + + def warning(self, message: str) -> None: + print(f" {self._c(_YELLOW + _BOLD, '⚠')} {self._c(_YELLOW, message)}") + + def success(self, message: str) -> None: + print(f" {self._c(_GREEN + _BOLD, '✓')} {self._c(_GREEN, message)}") + + def help(self, commands: dict[str, str]) -> None: + print() + print(f" {self._c(self.accent + _BOLD, 'Commands')}") + print(f" {self._c(_DARK_GRAY, _H_LINE * 8)}") + width = max((len(command) for command in commands), default=0) + for command, description in commands.items(): + print(f"{self._c(self.accent, f' {command:<{width}}')} {self._c(_GRAY, description)}") + print() + + def print_goodbye(self) -> None: + print(f"\n {self._c(_CYAN, _ICON_SMALL)} {self._c(_GRAY, 'Goodbye!')}\n") + + def create_prompt_session(self): + try: + from prompt_toolkit import PromptSession + from prompt_toolkit.auto_suggest import AutoSuggestFromHistory + from prompt_toolkit.history import FileHistory + except ImportError: + return None + + return PromptSession( + history=FileHistory(self.history_file), + auto_suggest=AutoSuggestFromHistory(), + style=self.get_prompt_style(), + enable_history_search=True, + ) + + def get_input(self, prompt_session, project_name: str = "", modified: bool = False, context: str = "") -> str: + if prompt_session is not None: + from prompt_toolkit.formatted_text import FormattedText + + return prompt_session.prompt(FormattedText(self.prompt_tokens(project_name, modified, context))).strip() + return input(self.prompt(project_name, modified, context)).strip() + + +_ANSI_256_TO_HEX = { + "\033[38;5;34m": "#00af00", + "\033[38;5;75m": "#5fafff", +} diff --git a/QGIS/agent-harness/pytest.ini b/QGIS/agent-harness/pytest.ini new file mode 100644 index 000000000..7b001b97a --- /dev/null +++ b/QGIS/agent-harness/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +addopts = --import-mode=importlib diff --git a/QGIS/agent-harness/setup.py b/QGIS/agent-harness/setup.py new file mode 100644 index 000000000..388cb4760 --- /dev/null +++ b/QGIS/agent-harness/setup.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""setup.py for cli-anything-qgis.""" + +from setuptools import find_namespace_packages, setup + +with open("cli_anything/qgis/README.md", "r", encoding="utf-8") as fh: + long_description = fh.read() + +setup( + name="cli-anything-qgis", + version="1.0.0", + author="cli-anything contributors", + author_email="", + description="CLI harness for QGIS using PyQGIS for project authoring and qgis_process for exports and processing.", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://github.com/HKUDS/CLI-Anything", + packages=find_namespace_packages(include=["cli_anything.*"]), + classifiers=[ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "Topic :: Software Development :: Libraries :: Python Modules", + "Topic :: Scientific/Engineering :: GIS", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + ], + python_requires=">=3.10", + install_requires=[ + "click>=8.0.0", + "prompt-toolkit>=3.0.0", + ], + extras_require={ + "dev": [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "jinja2>=3.1.0", + ], + }, + entry_points={ + "console_scripts": [ + "cli-anything-qgis=cli_anything.qgis.qgis_cli:main", + ], + }, + package_data={ + "cli_anything.qgis": ["skills/*.md"], + }, + include_package_data=True, + zip_safe=False, +)