Add QGIS agent harness

This commit is contained in:
scially
2026-04-09 18:21:06 +08:00
parent c05cfe1cf2
commit 9eb8bc5109
23 changed files with 3723 additions and 0 deletions
+4
View File
@@ -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/
+127
View File
@@ -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 <id>`)
- generic algorithm execution (`run <id>`)
- 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:
- `<project>.qgz` -> `<project>_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.
@@ -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
```
@@ -0,0 +1,3 @@
"""cli-anything-qgis package."""
__version__ = "1.0.0"
@@ -0,0 +1,7 @@
"""Run cli-anything-qgis as a module."""
from cli_anything.qgis.qgis_cli import main
if __name__ == "__main__":
main()
@@ -0,0 +1 @@
"""Core domain modules for cli-anything-qgis."""
@@ -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", []),
}
@@ -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,
}
@@ -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
@@ -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)
@@ -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", []),
}
@@ -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()
@@ -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", [])]
@@ -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()
@@ -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 <path>` 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
```
@@ -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 ========================
```
@@ -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
@@ -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}")
@@ -0,0 +1 @@
"""Utility helpers for cli-anything-qgis."""
@@ -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,
)
@@ -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",
}
+2
View File
@@ -0,0 +1,2 @@
[pytest]
addopts = --import-mode=importlib
+52
View File
@@ -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,
)