Initial garden layer package and tests

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+.pytest_cache/
+.venv/
+*.pyc

README.md

@@ -0,0 +1,45 @@
+# Garden Layer helper module
+
+The garden layer sits on top of `virtuoso_mcp`. It knows your garden-specific URIs (cycles, clone roots, seed products) but delegates every SPARQL/insert/discovery call to the MCP helpers.
+
+## Structure
+
+- `config.py` loads `.env` values and exposes the Garden URIs (seed product, cycle `2026-3`, clone property, etc.).
+- `helpers.py` defines `GardenLayer`, which wraps MCP tools such as `insert_triple`, `traverse_property`, and ontology discovery into domain actions (`add_seedling`, `traverse`).
+
+## Domain helpers
+
+`GardenLayer` now exposes these convenience methods:
+
+- `describe_subject(subject_uri, limit=10)` – returns predicates/objects (with labels) for a given node.
+- `path_traverse(subject_uri, property_path, direction="outgoing", limit=10)` – walks a property sequence from the subject and returns each step’s bindings.
+- `property_usage_statistics(property_uri, examples_limit=5)` – reports how often a property is used and includes sample subjects/objects.
+- `batch_insert(ttl, graph=None)` – pushes a TTL snippet (or multiple triples via TTL) into the chosen graph in a single request.
+
+These helpers keep domain code focused on planned workflows while still leveraging the generic MCP toolset.
+
+## Architectural split
+
+- `virtuoso_mcp` owns **generic** capabilities: query guardrails, traversal, ontology/class/property discovery.
+- `garden_layer` owns **domain workflows**: breeding lifecycle helpers, documentation flows, trait-specific routines.
+
+This split allows additional specialized layers to reuse the same generic ontology tooling without copy/paste.
+
+## Example
+
+```python
+from garden_layer import GardenLayer
+from garden_layer.config import SEED_PRODUCT, CYCLE_2026_3
+
+garden = GardenLayer()
+garden.add_seedling(
+    plant_uri="http://world.eu.org/example1#Plant_cookie_kerosene_2027",
+    seed_product_uri=SEED_PRODUCT,
+    cycle_uri=CYCLE_2026_3,
+    label="Cookie x Kerosene 2027",
+)
+``` 
+
+## Testing
+
+`garden_layer/test_garden_layer.py` exercises the garden helpers (`traverse`, `describe_subject`, `path_traverse`, `property_usage_statistics`, and `batch_insert`) against a running MCP server. Start the MCP service (`./virtuoso_mcp/run.sh` or `./restart.sh`) before running the tests and invoke them with `python3 -m pytest garden_layer/test_garden_layer.py`. If `pytest` is not installed, install it inside a virtual environment (e.g., `python3 -m venv .env && .env/bin/pip install pytest`).

__init__.py

@@ -0,0 +1,3 @@
+from .helpers import GardenLayer
+
+__all__ = ["GardenLayer"]

config.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+import os
+
+from dotenv import load_dotenv
+
+BASE_DIR = Path(__file__).resolve().parent
+load_dotenv(BASE_DIR.parent / ".env")
+
+MCP_URL = os.getenv("GARDEN_MCP_URL", "http://127.0.0.1:8501/mcp")
+GRAPH = os.getenv("GARDEN_GRAPH", "http://world.eu.org/example1")
+CLONE_OF = "http://world.eu.org/cannabis-breeding#cloneOf"
+IN_CYCLE = "http://world.eu.org/cannabis-breeding#inCycle"
+SEED_PRODUCT = "http://world.eu.org/example1#SeedProduct_7b71b88b-17f"
+CYCLE_2026_3 = "http://world.eu.org/example1#ProductionCycle_21d96b6c-179"
+KEROSENE_ROOT = "http://world.eu.org/example1#Plant_90d53925-bb5"

helpers.py

@@ -0,0 +1,85 @@
+from typing import Any, Dict, List, Optional
+
+import requests
+
+from .config import GRAPH, MCP_URL
+
+
+class GardenLayer:
+    """Domain helpers that orchestrate MCP calls for your garden project."""
+
+    def __init__(self, mcp_url: str = MCP_URL):
+        self.mcp_url = mcp_url
+
+    def _call(self, tool: str, payload: Dict[str, Any]) -> Any:
+        response = requests.post(self.mcp_url, json={"tool": tool, "input": payload}, timeout=10)
+        response.raise_for_status()
+        body = response.json()
+        if body.get("status") != "ok":
+            raise RuntimeError(body.get("detail", "MCP tool failed"))
+        return body["result"]
+
+    def traverse(self, subject_uri: str, property_uri: str, direction: str = "outgoing", limit: int = 20) -> Any:
+        return self._call("traverse_property", {
+            "subject_uri": subject_uri,
+            "property_uri": property_uri,
+            "direction": direction,
+            "limit": limit,
+        })
+
+    def describe_subject(self, subject_uri: str, limit: int = 10) -> Any:
+        return self._call("describe_subject", {
+            "subject_uri": subject_uri,
+            "limit": limit,
+        })
+
+    def path_traverse(
+        self,
+        subject_uri: str,
+        property_path: List[str],
+        direction: str = "outgoing",
+        limit: int = 10,
+    ) -> Any:
+        return self._call("path_traverse", {
+            "subject_uri": subject_uri,
+            "property_path": property_path,
+            "direction": direction,
+            "limit": limit,
+        })
+
+    def property_usage_statistics(self, property_uri: str, examples_limit: int = 5) -> Any:
+        return self._call("property_usage_statistics", {
+            "property_uri": property_uri,
+            "examples_limit": examples_limit,
+        })
+
+    def batch_insert(self, ttl: str, graph: Optional[str] = None) -> Any:
+        payload = {"ttl": ttl}
+        if graph:
+            payload["graph"] = graph
+        return self._call("batch_insert", payload)
+
+    def add_seedling(
+        self,
+        plant_uri: str,
+        seed_product_uri: str,
+        cycle_uri: str,
+        label: str,
+        strain_uri: str = "http://world.eu.org/example1#Strain_lucky_experimental_line",
+        graph: str = GRAPH,
+    ) -> None:
+        triples = [
+            ("http://www.w3.org/1999/02/22-rdf-syntax-ns#type", "http://world.eu.org/cannabis-breeding#IndividualPlant", "uri"),
+            ("http://www.w3.org/2000/01/rdf-schema#label", label, "literal"),
+            ("http://world.eu.org/cannabis-breeding#inCycle", cycle_uri, "uri"),
+            ("http://world.eu.org/cannabis-breeding#grownFromSeedProduct", seed_product_uri, "uri"),
+            ("http://world.eu.org/cannabis-breeding#belongsToStrain", strain_uri, "uri"),
+        ]
+        for predicate, obj, obj_type in triples:
+            self._call("insert_triple", {
+                "subject": plant_uri,
+                "predicate": predicate,
+                "object": obj,
+                "object_type": obj_type,
+                "graph": graph,
+            })

pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "garden-layer"
+version = "0.1.0"
+description = "Garden-specific helpers that orchestrate MCP calls for your breeding workflows."
+readme = "README.md"
+authors = [ { name = "Lukas Goldschmidt" } ]
+requires-python = ">=3.11"
+dependencies = [
+  "requests>=2.31",
+  "python-dotenv>=1.0",
+]
+
+[project.optional-dependencies]
+test = [
+  "pytest>=8.4",
+]
+
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"

requirements.txt

@@ -0,0 +1,3 @@
+requests>=2.31
+python-dotenv>=1.0
+pytest>=8.4

test.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ -n "${GARDEN_LAYER_VENV:-}" ]]; then
+  if [[ -f "$GARDEN_LAYER_VENV/bin/activate" ]]; then
+    # shellcheck disable=SC1091
+    source "$GARDEN_LAYER_VENV/bin/activate"
+  else
+    echo "Warning: GARDEN_LAYER_VENV is set to '$GARDEN_LAYER_VENV' but the activate script is missing." >&2
+  fi
+fi
+
+python3 -m pytest -v -s test_garden_layer.py

test_garden_layer.py

@@ -0,0 +1,79 @@
+import pytest
+import requests
+
+from garden_layer import GardenLayer
+from garden_layer.config import CLONE_OF, GRAPH, KEROSENE_ROOT
+
+
+def pytest_report_header(config):
+    return "Garden layer tests: requires virtuoso_mcp running at GardenLayer.MCP_URL"
+
+
+@pytest.fixture
+def garden_layer():
+    layer = GardenLayer()
+    try:
+        layer.traverse(KEROSENE_ROOT, CLONE_OF, direction="incoming", limit=1)
+    except requests.RequestException as exc:
+        pytest.skip(f"Cannot reach MCP server: {exc}")
+    return layer
+
+
+def test_traverse_clone_tree(garden_layer):
+    """Walk `cloneOf` incoming edges and ensure the root has clones to inspect."""
+    result = garden_layer.traverse(KEROSENE_ROOT, CLONE_OF, direction="incoming", limit=5)
+    assert isinstance(result, dict)
+    bindings = result.get("results", {}).get("bindings", [])
+    assert isinstance(bindings, list)
+    assert bindings, "Expected at least one clone binding"
+    print(f"Traverse of {KEROSENE_ROOT} returned {len(bindings)} clone candidates.")
+
+
+def test_describe_subject(garden_layer):
+    """Describe the root plant and surface the top predicates the garden helpers can reuse."""
+    summary = garden_layer.describe_subject(KEROSENE_ROOT, limit=5)
+    assert isinstance(summary, dict)
+    bindings = summary.get("results", {}).get("bindings", [])
+    assert bindings, "Expected describe_subject to return bindings"
+    triples = [f"{b['predicate']['value']} -> {b['object'].get('value', '<literal>')}" for b in bindings]
+    print("describe_subject returned:")
+    for triple in triples:
+        print("  ", triple)
+
+
+def test_path_traverse_lineage(garden_layer):
+    """Use the path helper to follow `cloneOf` and observe the lineage step."""
+    path = garden_layer.path_traverse(KEROSENE_ROOT, [CLONE_OF], direction="incoming", limit=5)
+    assert isinstance(path, dict)
+    bindings = path.get("result", {}).get("results", {}).get("bindings", [])
+    assert isinstance(bindings, list)
+    assert bindings, "Path traverse should find at least one step"
+    print(f"Property path {CLONE_OF} produced {len(bindings)} step bindings.")
+
+
+def test_property_usage_statistics(garden_layer):
+    """Summarize how the `cloneOf` property is used so future helpers can rely on frequency data."""
+    stats = garden_layer.property_usage_statistics(CLONE_OF, examples_limit=3)
+    assert isinstance(stats, dict)
+    count_bindings = stats.get("count", {}).get("results", {}).get("bindings", [])
+    assert count_bindings, "Expected usage count bindings"
+    usage_count = count_bindings[0].get("usageCount", {}).get("value")
+    example_bindings = stats.get("examples", {}).get("results", {}).get("bindings", [])
+    assert isinstance(example_bindings, list)
+    print(f"cloneOf usage count: {usage_count}")
+    print("example bindings:")
+    for binding in example_bindings:
+        subject = binding.get("subjectLabel", {}).get("value") or binding.get("subject", {}).get("value")
+        object_ = binding.get("objectLabel", {}).get("value") or binding.get("object", {}).get("value")
+        print(f"  - {subject} -> {object_}")
+
+
+def test_batch_insert(garden_layer):
+    """Batch-insert a TTL snippet and verify the query shape"""
+    ttl = '<http://world.eu.org/example1#batch_test_subject> <http://www.w3.org/2000/01/rdf-schema#label> "garden batch" .'
+    result = garden_layer.batch_insert(ttl=ttl, graph=GRAPH)
+    assert isinstance(result, dict)
+    assert "query" in result
+    assert "INSERT DATA" in result.get("query", "").upper()
+    print("batch_insert generated:")
+    print(result.get("query", ""))