Add gNMI-style JSON remediation rendering (#287)

CHANGELOG.md

@@ -70,6 +70,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `nc:operation="delete"` elements (keyed list entries delete by their key
   leaf, resolved against the running config), additions use the default merge
   operation, and attribute-level changes raise `InvalidConfigError`.
+- gNMI-style JSON remediation rendering (#287):
+  `WorkflowRemediation.remediation_json()` (and
+  `hier_config.formats.hconfig_to_gnmi_json()`) render a remediation between
+  `HConfig.from_json()` trees as a gNMI-SetRequest-style structure — added
+  and changed values render into an `update` object (modified keyed list
+  entries keep their identity leaf), negations become xpath-ish `delete`
+  paths with `[key=value]` selectors resolved against the running config,
+  and attribute-level changes raise `InvalidConfigError`.
 
 ### Fixed
 

docs/dev/architecture.md

@@ -149,8 +149,9 @@ The formats module maps JSON (e.g. OpenConfig) and XML (e.g. NETCONF payloads) o
 - `hconfig_from_json` / `hconfig_to_json` — invertible JSON mapping (keyed lists identified via `list_keys`).
 - `hconfig_from_xml` / `hconfig_to_xml` — invertible XML mapping (attributes and text content become specially-encoded leaves).
 - `hconfig_to_netconf_xml` — renders a remediation between `from_xml` trees as a NETCONF `edit-config` payload (deletions become `nc:operation="delete"` elements).
+- `hconfig_to_gnmi_json` — renders a remediation between `from_json` trees as a gNMI-SetRequest-style dict (additions render into an `update` object, deletions become xpath-ish paths with `[key=value]` selectors).
 
-These are exposed on `HConfig` as `from_json` / `from_xml` / `to_json` / `to_xml`, and on `WorkflowRemediation` as `remediation_netconf_xml()`. See [Loading Configurations](../user/loading-configs.md) for the mapping rules.
+These are exposed on `HConfig` as `from_json` / `from_xml` / `to_json` / `to_xml`, and on `WorkflowRemediation` as `remediation_netconf_xml()` / `remediation_json()`. See [Loading Configurations](../user/loading-configs.md) for the mapping rules.
 
 ---
 

docs/user/remediation-workflows.md

@@ -202,6 +202,20 @@ payload = wfr.remediation_netconf_xml()
 
 Deletions become elements with `nc:operation="delete"`; additions use the NETCONF default merge operation. Keyed list-entry deletions are expressed by their key leaf, resolved against the running config — pass `list_keys=` if your data does not use the default `name`/`id` keys. Attribute-level changes cannot be expressed as NETCONF operations and raise `InvalidConfigError`.
 
+## gNMI-style JSON remediation payloads
+
+When both configurations were built with [`HConfig.from_json()`](loading-configs.md#structured-formats-json-and-xml), the remediation can be rendered as a gNMI-SetRequest-style dict of update and delete sets:
+
+```python
+result = wfr.remediation_json()
+# {
+#     "update": {"system": {"config": {"hostname": "new"}}},
+#     "delete": ["interfaces/interface[name=eth1]"],
+# }
+```
+
+Added and changed values render into the `update` object using the same JSON mapping as `to_json()` (a modified keyed list entry keeps its identity leaf, so the update stays valid OpenConfig). Deletions become xpath-ish paths: keyed list entries get a `[key=value]` selector resolved against the running config — pass `list_keys=` if your data does not use the default `name`/`id` keys — while scalar leaves delete by their bare path (e.g. `system/config/hostname`). Backslashes and `]` inside selector values are escaped with a backslash. Element names themselves are not escaped, so keys containing `/` or `[` produce ambiguous paths.
+
 ## Next steps
 
 - [Working with Tags](tags.md) — filter the remediation for phased deployment.

hier_config/formats.py

@@ -38,14 +38,20 @@
 ``nc:operation="delete"`` elements; additions use the default merge
 operation). Attribute-level changes cannot be expressed as NETCONF
 operations and raise ``InvalidConfigError``.
+
+Remediation between ``hconfig_from_json`` trees can be rendered as a
+gNMI-SetRequest-style structure via ``hconfig_to_gnmi_json`` (deletions
+become xpath-ish paths with ``[key=value]`` selectors for keyed list
+entries; additions render into an ``update`` object using the JSON
+mapping above).
 """
 
 from __future__ import annotations
 
 import xml.etree.ElementTree as ET  # ruff:ignore[suspicious-xml-etree-import]
 from collections import Counter
 from json import JSONDecodeError, dumps, loads
-from typing import TYPE_CHECKING, Any, TypeAlias, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, TypeAlias, TypedDict, cast
 
 from .exceptions import InvalidConfigError
 from .registry import resolve_driver
@@ -66,6 +72,13 @@
 )
 
 
+class GnmiRemediation(TypedDict):
+    """gNMI-SetRequest-style remediation: an update tree and delete paths."""
+
+    update: dict[str, JsonValue]
+    delete: list[str]
+
+
 def hconfig_from_json(
     platform_or_driver: Platform | str | HConfigDriverBase,
     data: str | dict[str, Any],
@@ -387,15 +400,153 @@ def _netconf_delete_element(
     if len(words) == 1:
         return element
     # A keyed list entry (branch in the running config) deletes by key leaf.
-    if (
-        running_parent is not None
-        and (running_entry := running_parent.get_child(equals=positive_text))
-        is not None
-        and running_entry.children
-    ):
-        for key in list_keys:
-            if running_entry.get_child(equals=f"{key} {words[1]}") is not None:
-                ET.SubElement(element, key).text = _xml_text(words[1])
-                return element
+    key = _running_entry_key(running_parent, positive_text, words[1], list_keys)
+    if key is not None:
+        ET.SubElement(element, key).text = _xml_text(words[1])
+        return element
     element.text = _xml_text(words[1])
     return element
+
+
+def _matching_list_key(
+    entry: HConfigBase,
+    raw_value: str,
+    list_keys: tuple[str, ...],
+) -> str | None:
+    for key in list_keys:
+        if entry.get_child(equals=f"{key} {raw_value}") is not None:
+            return key
+    return None
+
+
+def _running_entry_key(
+    running_parent: HConfigBase | None,
+    positive_text: str,
+    raw_value: str,
+    list_keys: tuple[str, ...],
+) -> str | None:
+    """Key leaf identifying `positive_text` as a keyed list entry, if any."""
+    if running_parent is None:
+        return None
+    running_entry = running_parent.get_child(equals=positive_text)
+    if running_entry is None or not running_entry.children:
+        return None
+    return _matching_list_key(running_entry, raw_value, list_keys)
+
+
+def hconfig_to_gnmi_json(
+    remediation: HConfig,
+    *,
+    running: HConfig | None = None,
+    list_keys: tuple[str, ...] | None = None,
+) -> GnmiRemediation:
+    """Render a remediation between `hconfig_from_json` trees as gNMI-style sets.
+
+    Negated nodes become xpath-ish delete paths; everything else renders
+    into the `update` object via the JSON mapping. When `running` is given,
+    deletions of keyed list entries get `[key=value]` selectors (keys found
+    via `list_keys`); without it, deletions fall back to bare leaf paths.
+    """
+    result: GnmiRemediation = {"update": {}, "delete": []}
+    context = _GnmiContext(
+        delete=result["delete"],
+        negation_prefix=remediation.driver.negation_prefix,
+        list_keys=list_keys or DEFAULT_LIST_KEYS,
+    )
+    _gnmi_into(remediation, result["update"], (), running, context)
+    return result
+
+
+class _GnmiContext(NamedTuple):
+    delete: list[str]
+    negation_prefix: str
+    list_keys: tuple[str, ...]
+
+
+def _gnmi_into(
+    node: HConfigBase,
+    update: dict[str, JsonValue],
+    path: tuple[str, ...],
+    running_node: HConfigBase | None,
+    context: _GnmiContext,
+) -> None:
+    for child in node.children:
+        if child.text.startswith(context.negation_prefix):
+            # A negated child is resolved against the parent's running node.
+            context.delete.append(
+                _gnmi_delete_path(
+                    path,
+                    child.text.removeprefix(context.negation_prefix),
+                    running_node,
+                    context.list_keys,
+                )
+            )
+            continue
+        words = child.text.split(maxsplit=1)
+        if not child.children:
+            value: JsonValue = _leaf_value(words[1]) if len(words) > 1 else {}
+            _store_json_member(update, words[0], value, force_list=False)
+            continue
+        running_child = (
+            running_node.get_child(equals=child.text) if running_node else None
+        )
+        segment = words[0]
+        key_name: str | None = None
+        if len(words) > 1:
+            key_name = _gnmi_identity_key(
+                child, running_child, words[1], context.list_keys
+            )
+            segment = (
+                f"{words[0]}[{key_name or context.list_keys[0]}"
+                f"={_gnmi_selector_value(words[1])}]"
+            )
+        child_update: dict[str, JsonValue] = {}
+        _gnmi_into(child, child_update, (*path, segment), running_child, context)
+        if not child_update:
+            # The branch contained only deletions.
+            continue
+        if key_name is not None and key_name not in child_update:
+            child_update = {key_name: _leaf_value(words[1]), **child_update}
+        _store_json_member(update, words[0], child_update, force_list=len(words) > 1)
+
+
+def _gnmi_identity_key(
+    entry: HConfigChild,
+    running_entry: HConfigBase | None,
+    raw_value: str,
+    list_keys: tuple[str, ...],
+) -> str | None:
+    for source in (entry, running_entry):
+        if source is None:
+            continue
+        key = _matching_list_key(source, raw_value, list_keys)
+        if key is not None:
+            return key
+    return None
+
+
+def _gnmi_selector_value(raw: str) -> str:
+    return _xml_text(raw).replace("\\", "\\\\").replace("]", "\\]")
+
+
+def _gnmi_delete_path(
+    parent_path: tuple[str, ...],
+    positive_text: str,
+    running_parent: HConfigBase | None,
+    list_keys: tuple[str, ...],
+) -> str:
+    words = positive_text.split(maxsplit=1)
+    if words[0].startswith("@"):
+        message = (
+            "Attribute changes cannot be expressed as gNMI delete paths:"
+            f" {positive_text!r}"
+        )
+        raise InvalidConfigError(message)
+    segment = words[0]
+    # A keyed list entry (branch in the running config) deletes by selector;
+    # a scalar leaf deletes by its bare path (the value is dropped).
+    if len(words) > 1:
+        key = _running_entry_key(running_parent, positive_text, words[1], list_keys)
+        if key is not None:
+            segment = f"{words[0]}[{key}={_gnmi_selector_value(words[1])}]"
+    return "/".join((*parent_path, segment))

hier_config/workflows.py

@@ -1,10 +1,17 @@
-from collections.abc import Callable, Iterable
+from __future__ import annotations
+
 from logging import getLogger
+from typing import TYPE_CHECKING
 
 from .exceptions import IncompatibleDriverError
-from .models import TagRule
 from .root import HConfig
 
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterable
+
+    from .formats import GnmiRemediation
+    from .models import TagRule
+
 logger = getLogger(__name__)
 
 
@@ -132,16 +139,35 @@ def remediation_netconf_xml(
         Keyed list-entry deletions are expressed by their key leaf, resolved
         against the running config via `list_keys`.
         """
-        from .formats import (
-            hconfig_to_netconf_xml,
-        )
+        from .formats import hconfig_to_netconf_xml
 
         return hconfig_to_netconf_xml(
             self.remediation_config,
             running=self.running_config,
             list_keys=list_keys,
         )
 
+    def remediation_json(
+        self,
+        *,
+        list_keys: tuple[str, ...] | None = None,
+    ) -> GnmiRemediation:
+        """Render the remediation as a gNMI-SetRequest-style dict.
+
+        Requires running and generated configs built by `HConfig.from_json()`.
+        Returns `{"update": ..., "delete": [...]}` — added/changed values as a
+        JSON tree and deletions as xpath-ish paths. Keyed list-entry deletions
+        get `[key=value]` selectors, resolved against the running config via
+        `list_keys`.
+        """
+        from .formats import hconfig_to_gnmi_json
+
+        return hconfig_to_gnmi_json(
+            self.remediation_config,
+            running=self.running_config,
+            list_keys=list_keys,
+        )
+
     def apply_remediation_tag_rules(self, tag_rules: tuple[TagRule, ...]) -> None:
         """Applies tag rules to selectively label parts of the remediation configuration.