merge: sync upstream/main and normalize lldb metadata

This commit is contained in:
aimidi
2026-04-17 16:56:04 +08:00
28 changed files with 4114 additions and 10 deletions
+4
View File
@@ -92,6 +92,7 @@
!/renderdoc/
!/cloudcompare/
!/openscreen/
!/QGIS/
!/n8n/
!/obsidian/
!/lldb/
@@ -165,6 +166,8 @@
/cloudcompare/.*
/openscreen/*
/openscreen/.*
/QGIS/*
/QGIS/.*
/cloudanalyzer/*
/cloudanalyzer/.*
/wiremock/*
@@ -220,6 +223,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 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,140 @@
"""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,
)
if not os.path.exists(output):
raise QgisBackendError(
f"Export succeeded but output file was not created: {output}"
)
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,
)
if not os.path.exists(output):
raise QgisBackendError(
f"Export succeeded but output file was not created: {output}"
)
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,195 @@
"""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."
)
if extent.width() == 0 and extent.height() == 0:
extent.grow(1.0)
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 functools
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."""
@functools.wraps(func)
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
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,500 @@
"""cli-anything REPL Skin — Unified terminal interface for all CLI harnesses.
Copy this file into your CLI package at:
cli_anything/<software>/utils/repl_skin.py
Usage:
from cli_anything.<software>.utils.repl_skin import ReplSkin
skin = ReplSkin("shotcut", version="1.0.0")
skin.print_banner()
prompt_text = skin.prompt(project_name="my_video.mlt", modified=True)
skin.success("Project saved")
skin.error("File not found")
skin.warning("Unsaved changes")
skin.info("Processing 24 clips...")
skin.status("Track 1", "3 clips, 00:02:30")
skin.table(headers, rows)
skin.print_goodbye()
"""
import os
import sys
# ── ANSI color codes (no external deps for core styling) ──────────────
_RESET = "\033[0m"
_BOLD = "\033[1m"
_DIM = "\033[2m"
_ITALIC = "\033[3m"
_UNDERLINE = "\033[4m"
# Brand colors
_CYAN = "\033[38;5;80m" # cli-anything brand cyan
_CYAN_BG = "\033[48;5;80m"
_WHITE = "\033[97m"
_GRAY = "\033[38;5;245m"
_DARK_GRAY = "\033[38;5;240m"
_LIGHT_GRAY = "\033[38;5;250m"
# Software accent colors — each software gets a unique accent
_ACCENT_COLORS = {
"gimp": "\033[38;5;214m", # warm orange
"blender": "\033[38;5;208m", # deep orange
"inkscape": "\033[38;5;39m", # bright blue
"audacity": "\033[38;5;33m", # navy blue
"libreoffice": "\033[38;5;40m", # green
"obs_studio": "\033[38;5;55m", # purple
"kdenlive": "\033[38;5;69m", # slate blue
"shotcut": "\033[38;5;35m", # teal green
"qgis": "\033[38;5;34m", # qgis green
}
_DEFAULT_ACCENT = "\033[38;5;75m" # default sky blue
# Status colors
_GREEN = "\033[38;5;78m"
_YELLOW = "\033[38;5;220m"
_RED = "\033[38;5;196m"
_BLUE = "\033[38;5;75m"
_MAGENTA = "\033[38;5;176m"
# ── Brand icon ────────────────────────────────────────────────────────
# The cli-anything icon: a small colored diamond/chevron mark
_ICON = f"{_CYAN}{_BOLD}◆{_RESET}"
_ICON_SMALL = f"{_CYAN}▸{_RESET}"
# ── Box drawing characters ────────────────────────────────────────────
_H_LINE = "─"
_V_LINE = "│"
_TL = "╭"
_TR = "╮"
_BL = "╰"
_BR = "╯"
_T_DOWN = "┬"
_T_UP = "┴"
_T_RIGHT = "├"
_T_LEFT = "┤"
_CROSS = "┼"
def _strip_ansi(text: str) -> str:
"""Remove ANSI escape codes for length calculation."""
import re
return re.sub(r"\033\[[^m]*m", "", text)
def _visible_len(text: str) -> int:
"""Get visible length of text (excluding ANSI codes)."""
return len(_strip_ansi(text))
class ReplSkin:
"""Unified REPL skin for cli-anything CLIs.
Provides consistent branding, prompts, and message formatting
across all CLI harnesses built with the cli-anything methodology.
"""
def __init__(self, software: str, version: str = "1.0.0",
history_file: str | None = None):
"""Initialize the REPL skin.
Args:
software: Software name (e.g., "gimp", "shotcut", "blender").
version: CLI version string.
history_file: Path for persistent command history.
Defaults to ~/.cli-anything-<software>/history
"""
self.software = software.lower().replace("-", "_")
self.display_name = software.replace("_", " ").title()
self.version = version
self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT)
# History file
if history_file is None:
from pathlib import Path
hist_dir = Path.home() / f".cli-anything-{self.software}"
hist_dir.mkdir(parents=True, exist_ok=True)
self.history_file = str(hist_dir / "history")
else:
self.history_file = history_file
# Detect terminal capabilities
self._color = self._detect_color_support()
def _detect_color_support(self) -> bool:
"""Check if terminal supports color."""
if os.environ.get("NO_COLOR"):
return False
if 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:
"""Apply color code if colors are supported."""
if not self._color:
return text
return f"{code}{text}{_RESET}"
# ── Banner ────────────────────────────────────────────────────────
def print_banner(self):
"""Print the startup banner with branding."""
inner = 54
def _box_line(content: str) -> str:
"""Wrap content in box drawing, padding to inner width."""
pad = inner - _visible_len(content)
vl = self._c(_DARK_GRAY, _V_LINE)
return f"{vl}{content}{' ' * max(0, pad)}{vl}"
top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}")
bot = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}")
# Title: ◆ cli-anything · Shotcut
icon = self._c(_CYAN + _BOLD, "◆")
brand = self._c(_CYAN + _BOLD, "cli-anything")
dot = self._c(_DARK_GRAY, "·")
name = self._c(self.accent + _BOLD, self.display_name)
title = f" {icon} {brand} {dot} {name}"
ver = f" {self._c(_DARK_GRAY, f' v{self.version}')}"
tip = f" {self._c(_DARK_GRAY, ' Type help for commands, quit to exit')}"
empty = ""
print(top)
print(_box_line(title))
print(_box_line(ver))
print(_box_line(empty))
print(_box_line(tip))
print(bot)
print()
# ── Prompt ────────────────────────────────────────────────────────
def prompt(self, project_name: str = "", modified: bool = False,
context: str = "") -> str:
"""Build a styled prompt string for prompt_toolkit or input().
Args:
project_name: Current project name (empty if none open).
modified: Whether the project has unsaved changes.
context: Optional extra context to show in prompt.
Returns:
Formatted prompt string.
"""
parts = []
# Icon
if self._color:
parts.append(f"{_CYAN}◆{_RESET} ")
else:
parts.append("> ")
# Software name
parts.append(self._c(self.accent + _BOLD, self.software))
# Project context
if project_name or context:
ctx = context or project_name
mod = "*" if modified else ""
parts.append(f" {self._c(_DARK_GRAY, '[')}")
parts.append(self._c(_LIGHT_GRAY, f"{ctx}{mod}"))
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 = ""):
"""Build prompt_toolkit formatted text tokens for the prompt.
Use with prompt_toolkit's FormattedText for proper ANSI handling.
Returns:
list of (style, text) tuples for prompt_toolkit.
"""
accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
tokens = []
tokens.append(("class:icon", "◆ "))
tokens.append(("class:software", self.software))
if project_name or context:
ctx = context or project_name
mod = "*" if modified else ""
tokens.append(("class:bracket", " ["))
tokens.append(("class:context", f"{ctx}{mod}"))
tokens.append(("class:bracket", "]"))
tokens.append(("class:arrow", " ❯ "))
return tokens
def get_prompt_style(self):
"""Get a prompt_toolkit Style object matching the skin.
Returns:
prompt_toolkit.styles.Style
"""
try:
from prompt_toolkit.styles import Style
except ImportError:
return None
accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
return Style.from_dict({
"icon": "#5fdfdf bold", # cyan brand color
"software": f"{accent_hex} bold",
"bracket": "#585858",
"context": "#bcbcbc",
"arrow": "#808080",
# Completion menu
"completion-menu.completion": "bg:#303030 #bcbcbc",
"completion-menu.completion.current": f"bg:{accent_hex} #000000",
"completion-menu.meta.completion": "bg:#303030 #808080",
"completion-menu.meta.completion.current": f"bg:{accent_hex} #000000",
# Auto-suggest
"auto-suggest": "#585858",
# Bottom toolbar
"bottom-toolbar": "bg:#1c1c1c #808080",
"bottom-toolbar.text": "#808080",
})
# ── Messages ──────────────────────────────────────────────────────
def success(self, message: str):
"""Print a success message with green checkmark."""
icon = self._c(_GREEN + _BOLD, "✓")
print(f" {icon} {self._c(_GREEN, message)}")
def error(self, message: str):
"""Print an error message with red cross."""
icon = self._c(_RED + _BOLD, "✗")
print(f" {icon} {self._c(_RED, message)}", file=sys.stderr)
def warning(self, message: str):
"""Print a warning message with yellow triangle."""
icon = self._c(_YELLOW + _BOLD, "⚠")
print(f" {icon} {self._c(_YELLOW, message)}")
def info(self, message: str):
"""Print an info message with blue dot."""
icon = self._c(_BLUE, "●")
print(f" {icon} {self._c(_LIGHT_GRAY, message)}")
def hint(self, message: str):
"""Print a subtle hint message."""
print(f" {self._c(_DARK_GRAY, message)}")
def section(self, title: str):
"""Print a section header."""
print()
print(f" {self._c(self.accent + _BOLD, title)}")
print(f" {self._c(_DARK_GRAY, _H_LINE * len(title))}")
# ── Status display ────────────────────────────────────────────────
def status(self, label: str, value: str):
"""Print a key-value status line."""
lbl = self._c(_GRAY, f" {label}:")
val = self._c(_WHITE, f" {value}")
print(f"{lbl}{val}")
def status_block(self, items: dict[str, str], title: str = ""):
"""Print a block of status key-value pairs.
Args:
items: Dict of label -> value pairs.
title: Optional title for the block.
"""
if title:
self.section(title)
max_key = max(len(k) for k in items) if items else 0
for label, value in items.items():
lbl = self._c(_GRAY, f" {label:<{max_key}}")
val = self._c(_WHITE, f" {value}")
print(f"{lbl}{val}")
def progress(self, current: int, total: int, label: str = ""):
"""Print a simple progress indicator.
Args:
current: Current step number.
total: Total number of steps.
label: Optional label for the progress.
"""
pct = int(current / total * 100) if total > 0 else 0
bar_width = 20
filled = int(bar_width * current / total) if total > 0 else 0
bar = "█" * filled + "░" * (bar_width - filled)
text = f" {self._c(_CYAN, bar)} {self._c(_GRAY, f'{pct:3d}%')}"
if label:
text += f" {self._c(_LIGHT_GRAY, label)}"
print(text)
# ── Table display ─────────────────────────────────────────────────
def table(self, headers: list[str], rows: list[list[str]],
max_col_width: int = 40):
"""Print a formatted table with box-drawing characters.
Args:
headers: Column header strings.
rows: List of rows, each a list of cell strings.
max_col_width: Maximum column width before truncation.
"""
if not headers:
return
# Calculate column widths
col_widths = [min(len(h), max_col_width) for h in headers]
for row in rows:
for i, cell in enumerate(row):
if i < len(col_widths):
col_widths[i] = min(
max(col_widths[i], len(str(cell))), max_col_width
)
def pad(text: str, width: int) -> str:
t = str(text)[:width]
return t + " " * (width - len(t))
# Header
header_cells = [
self._c(_CYAN + _BOLD, pad(h, col_widths[i]))
for i, h in enumerate(headers)
]
sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
header_line = f" {sep.join(header_cells)}"
print(header_line)
# Separator
sep_parts = [self._c(_DARK_GRAY, _H_LINE * w) for w in col_widths]
sep_line = self._c(_DARK_GRAY, f" {'───'.join([_H_LINE * w for w in col_widths])}")
print(sep_line)
# Rows
for row in rows:
cells = []
for i, cell in enumerate(row):
if i < len(col_widths):
cells.append(self._c(_LIGHT_GRAY, pad(str(cell), col_widths[i])))
row_sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
print(f" {row_sep.join(cells)}")
# ── Help display ──────────────────────────────────────────────────
def help(self, commands: dict[str, str]):
"""Print a formatted help listing.
Args:
commands: Dict of command -> description pairs.
"""
self.section("Commands")
max_cmd = max(len(c) for c in commands) if commands else 0
for cmd, desc in commands.items():
cmd_styled = self._c(self.accent, f" {cmd:<{max_cmd}}")
desc_styled = self._c(_GRAY, f" {desc}")
print(f"{cmd_styled}{desc_styled}")
print()
# ── Goodbye ───────────────────────────────────────────────────────
def print_goodbye(self):
"""Print a styled goodbye message."""
print(f"\n {_ICON_SMALL} {self._c(_GRAY, 'Goodbye!')}\n")
# ── Prompt toolkit session factory ────────────────────────────────
def create_prompt_session(self):
"""Create a prompt_toolkit PromptSession with skin styling.
Returns:
A configured PromptSession, or None if prompt_toolkit unavailable.
"""
try:
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.formatted_text import FormattedText
style = self.get_prompt_style()
session = PromptSession(
history=FileHistory(self.history_file),
auto_suggest=AutoSuggestFromHistory(),
style=style,
enable_history_search=True,
)
return session
except ImportError:
return None
def get_input(self, pt_session, project_name: str = "",
modified: bool = False, context: str = "") -> str:
"""Get input from user using prompt_toolkit or fallback.
Args:
pt_session: A prompt_toolkit PromptSession (or None).
project_name: Current project name.
modified: Whether project has unsaved changes.
context: Optional context string.
Returns:
User input string (stripped).
"""
if pt_session is not None:
from prompt_toolkit.formatted_text import FormattedText
tokens = self.prompt_tokens(project_name, modified, context)
return pt_session.prompt(FormattedText(tokens)).strip()
else:
raw_prompt = self.prompt(project_name, modified, context)
return input(raw_prompt).strip()
# ── Toolbar builder ───────────────────────────────────────────────
def bottom_toolbar(self, items: dict[str, str]):
"""Create a bottom toolbar callback for prompt_toolkit.
Args:
items: Dict of label -> value pairs to show in toolbar.
Returns:
A callable that returns FormattedText for the toolbar.
"""
def toolbar():
from prompt_toolkit.formatted_text import FormattedText
parts = []
for i, (k, v) in enumerate(items.items()):
if i > 0:
parts.append(("class:bottom-toolbar.text", " │ "))
parts.append(("class:bottom-toolbar.text", f" {k}: "))
parts.append(("class:bottom-toolbar", v))
return FormattedText(parts)
return toolbar
# ── ANSI 256-color to hex mapping (for prompt_toolkit styles) ─────────
_ANSI_256_TO_HEX = {
"\033[38;5;33m": "#0087ff", # audacity navy blue
"\033[38;5;34m": "#00af00", # qgis green
"\033[38;5;35m": "#00af5f", # shotcut teal
"\033[38;5;39m": "#00afff", # inkscape bright blue
"\033[38;5;40m": "#00d700", # libreoffice green
"\033[38;5;55m": "#5f00af", # obs purple
"\033[38;5;69m": "#5f87ff", # kdenlive slate blue
"\033[38;5;75m": "#5fafff", # default sky blue
"\033[38;5;80m": "#5fd7d7", # brand cyan
"\033[38;5;208m": "#ff8700", # blender deep orange
"\033[38;5;214m": "#ffaf00", # gimp warm orange
}
+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,
)
+10 -3
View File
@@ -980,12 +980,19 @@ Each application received complete, production-ready CLI interfaces — not demo
<td align="center">✅ 14</td>
</tr>
<tr>
<td align="center"><strong>🗺️ <a href="QGIS/agent-harness/">QGIS</a></strong></td>
<td>Geospatial Analysis &amp; Mapping</td>
<td><code>cli-anything-qgis</code></td>
<td>PyQGIS + qgis_process</td>
<td align="center">✅ 22</td>
</tr>
<tr>
<td align="center" colspan="4"><strong>Total</strong></td>
<td align="center"><strong>✅ 2,159</strong></td>
<td align="center"><strong>✅ 2,173</strong></td>
</tr>
</table>
> **100% pass rate** across all 2,159 tests — 1,570 unit tests + 570 end-to-end tests + 19 Node.js tests.
> **100% pass rate** across all 2,173 tests — 1,582 unit tests + 572 end-to-end tests + 19 Node.js tests.
---
@@ -1027,7 +1034,7 @@ cloudcompare 88 passed ✅ (49 unit + 39 e2e)
openscreen 101 passed ✅ (78 unit + 23 e2e)
cloudanalyzer 14 passed ✅ (7 unit + 7 e2e)
──────────────────────────────────────────────────────────────────────────────
TOTAL 2,159 passed ✅ 100% pass rate
TOTAL 2,173 passed ✅ 100% pass rate
```
---
+4 -2
View File
@@ -159,14 +159,16 @@ def extract_system_package(content: str) -> Optional[str]:
patterns = [
r"`apt install ([\w\-]+)`",
r"`brew install ([\w\-]+)`",
r"apt-get install ([\w\-]+)",
r"`apt-get install ([\w\-]+)`",
]
for pattern in patterns:
match = re.search(pattern, content)
if match:
package = match.group(1)
if "apt" in pattern:
if "apt-get" in pattern:
return f"apt-get install {package}"
elif "apt" in pattern:
return f"apt install {package}"
elif "brew" in pattern:
return f"brew install {package}"
@@ -30,6 +30,7 @@ from skill_generator import (
generate_skill_md_simple,
generate_skill_file,
extract_intro_from_readme,
extract_system_package,
extract_version_from_setup,
SkillMetadata,
CommandInfo,
@@ -206,6 +207,35 @@ class TestExtractIntroFromReadme:
assert "Line two" in intro
# ─── extract_system_package Tests ─────────────────────
class TestExtractSystemPackage:
def test_apt_install(self):
content = "Install with `apt install mytool`."
result = extract_system_package(content)
assert result == "apt install mytool"
def test_brew_install(self):
content = "Install with `brew install mytool`."
result = extract_system_package(content)
assert result == "brew install mytool"
def test_apt_get_install_returns_apt_get_command(self):
# Regression: apt-get pattern contains "apt" as a substring, so the
# condition must check "apt-get" before "apt" to avoid returning the
# wrong command ("apt install" instead of "apt-get install").
content = "Install with `apt-get install mytool`."
result = extract_system_package(content)
assert result == "apt-get install mytool", (
f"Expected 'apt-get install mytool', got {result!r}"
)
def test_returns_none_when_no_match(self):
content = "No installation instructions here."
assert extract_system_package(content) is None
# ─── generate_skill_md Tests ───────────────────────────────────────────
+5 -3
View File
@@ -76,16 +76,18 @@ def extract_system_package(content: str) -> Optional[str]:
patterns = [
r"`apt install ([\w\-]+)`",
r"`brew install ([\w\-]+)`",
r"apt-get install ([\w\-]+)",
r"`apt-get install ([\w\-]+)`",
]
for pattern in patterns:
match = re.search(pattern, content)
if match:
package = match.group(1)
if "apt" in pattern:
if "apt-get" in pattern:
return f"apt-get install {package}"
elif "apt" in pattern:
return f"apt install {package}"
if "brew" in pattern:
elif "brew" in pattern:
return f"brew install {package}"
return None
+21 -2
View File
@@ -278,12 +278,17 @@
"description": "Stateful native debugging via the LLDB Python API with JSON-friendly inspection commands",
"requires": "LLDB installation with Python bindings available (for example LLVM.LLVM on Windows)",
"homepage": "https://lldb.llvm.org",
"source_url": null,
"install_cmd": "pip install git+https://github.com/HKUDS/CLI-Anything.git#subdirectory=lldb/agent-harness",
"entry_point": "cli-anything-lldb",
"skill_md": "lldb/agent-harness/cli_anything/lldb/skills/SKILL.md",
"category": "debugging",
"contributor": "AiMiDi",
"contributor_url": "https://github.com/AiMiDi"
"contributors": [
{
"name": "AiMiDi",
"url": "https://github.com/AiMiDi"
}
]
},
{
"name": "zotero",
@@ -517,6 +522,20 @@
}
]
},
{
"name": "qgis",
"display_name": "QGIS",
"version": "1.0.0",
"description": "Geospatial project authoring, layout export, and processing via PyQGIS and qgis_process",
"requires": "QGIS with qgis_process on PATH, PyQGIS importable",
"homepage": "https://qgis.org",
"install_cmd": "pip install git+https://github.com/HKUDS/CLI-Anything.git#subdirectory=QGIS/agent-harness",
"entry_point": "cli-anything-qgis",
"skill_md": "QGIS/agent-harness/cli_anything/qgis/skills/SKILL.md",
"category": "scientific",
"contributor": "scially",
"contributor_url": "https://github.com/scially"
},
{
"name": "chromadb",
"display_name": "ChromaDB",