Add json_attributes_path configuration for command_line sensor (home-…

…assistant#116656)

Add json attributes path config to command line sensor

homeassistant/components/command_line/__init__.py

@@ -60,12 +60,17 @@
 from homeassistant.helpers.trigger_template_entity import CONF_AVAILABILITY
 from homeassistant.helpers.typing import ConfigType
 
-from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
+from .const import (
+    CONF_COMMAND_TIMEOUT,
+    CONF_JSON_ATTRIBUTES,
+    CONF_JSON_ATTRIBUTES_PATH,
+    DEFAULT_TIMEOUT,
+    DOMAIN,
+)
 
 BINARY_SENSOR_DEFAULT_NAME = "Binary Command Sensor"
 DEFAULT_PAYLOAD_ON = "ON"
 DEFAULT_PAYLOAD_OFF = "OFF"
-CONF_JSON_ATTRIBUTES = "json_attributes"
 SENSOR_DEFAULT_NAME = "Command Sensor"
 CONF_NOTIFIERS = "notifiers"
 
@@ -126,6 +131,7 @@
         vol.Required(CONF_COMMAND): cv.string,
         vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
         vol.Optional(CONF_JSON_ATTRIBUTES): cv.ensure_list_csv,
+        vol.Optional(CONF_JSON_ATTRIBUTES_PATH): cv.string,
         vol.Optional(CONF_NAME, default=SENSOR_DEFAULT_NAME): cv.string,
         vol.Optional(CONF_ICON): cv.template,
         vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,

homeassistant/components/command_line/const.py

@@ -18,6 +18,8 @@
 LOGGER = logging.getLogger(__package__)
 
 CONF_COMMAND_TIMEOUT = "command_timeout"
+CONF_JSON_ATTRIBUTES = "json_attributes"
+CONF_JSON_ATTRIBUTES_PATH = "json_attributes_path"
 DEFAULT_TIMEOUT = 15
 DOMAIN = "command_line"
 PLATFORMS = [

homeassistant/components/command_line/manifest.json

@@ -3,5 +3,6 @@
   "name": "Command Line",
   "codeowners": ["@gjohansson-ST"],
   "documentation": "https://www.home-assistant.io/integrations/command_line",
-  "iot_class": "local_polling"
+  "iot_class": "local_polling",
+  "requirements": ["jsonpath==0.82.2"]
 }

homeassistant/components/command_line/sensor.py

@@ -8,6 +8,8 @@
 import json
 from typing import Any, cast
 
+from jsonpath import jsonpath
+
 from homeassistant.components.sensor import SensorDeviceClass
 from homeassistant.components.sensor.helpers import async_parse_date_datetime
 from homeassistant.const import (
@@ -25,11 +27,15 @@
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 from homeassistant.util import dt as dt_util
 
-from .const import CONF_COMMAND_TIMEOUT, LOGGER, TRIGGER_ENTITY_OPTIONS
+from .const import (
+    CONF_COMMAND_TIMEOUT,
+    CONF_JSON_ATTRIBUTES,
+    CONF_JSON_ATTRIBUTES_PATH,
+    LOGGER,
+    TRIGGER_ENTITY_OPTIONS,
+)
 from .utils import async_check_output_or_log
 
-CONF_JSON_ATTRIBUTES = "json_attributes"
-
 DEFAULT_NAME = "Command Sensor"
 
 SCAN_INTERVAL = timedelta(seconds=60)
@@ -49,6 +55,7 @@ async def async_setup_platform(
     command: str = sensor_config[CONF_COMMAND]
     command_timeout: int = sensor_config[CONF_COMMAND_TIMEOUT]
     json_attributes: list[str] | None = sensor_config.get(CONF_JSON_ATTRIBUTES)
+    json_attributes_path: str | None = sensor_config.get(CONF_JSON_ATTRIBUTES_PATH)
     scan_interval: timedelta = sensor_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
     data = CommandSensorData(hass, command, command_timeout)
 
@@ -67,6 +74,7 @@ async def async_setup_platform(
                 trigger_entity_config,
                 value_template,
                 json_attributes,
+                json_attributes_path,
                 scan_interval,
             )
         ]
@@ -84,13 +92,15 @@ def __init__(
         config: ConfigType,
         value_template: Template | None,
         json_attributes: list[str] | None,
+        json_attributes_path: str | None,
         scan_interval: timedelta,
     ) -> None:
         """Initialize the sensor."""
         super().__init__(self.hass, config)
         self.data = data
         self._attr_extra_state_attributes: dict[str, Any] = {}
         self._json_attributes = json_attributes
+        self._json_attributes_path = json_attributes_path
         self._attr_native_value = None
         self._value_template = value_template
         self._scan_interval = scan_interval
@@ -141,6 +151,13 @@ async def _async_update(self) -> None:
             if value:
                 try:
                     json_dict = json.loads(value)
+                    if self._json_attributes_path is not None:
+                        json_dict = jsonpath(json_dict, self._json_attributes_path)
+                    # jsonpath will always store the result in json_dict[0]
+                    # so the next line happens to work exactly as needed to
+                    # find the result
+                    if isinstance(json_dict, list):
+                        json_dict = json_dict[0]
                     if isinstance(json_dict, Mapping):
                         self._attr_extra_state_attributes = {
                             k: json_dict[k]

requirements_all.txt

@@ -1187,6 +1187,7 @@ jaraco.abode==5.1.2
 # homeassistant.components.jellyfin
 jellyfin-apiclient-python==1.9.2
 
+# homeassistant.components.command_line
 # homeassistant.components.rest
 jsonpath==0.82.2
 

requirements_test_all.txt

@@ -974,6 +974,7 @@ jaraco.abode==5.1.2
 # homeassistant.components.jellyfin
 jellyfin-apiclient-python==1.9.2
 
+# homeassistant.components.command_line
 # homeassistant.components.rest
 jsonpath==0.82.2
 

tests/components/command_line/test_sensor.py

@@ -467,6 +467,46 @@ async def test_update_with_unnecessary_json_attrs(
     assert "key_three" not in entity_state.attributes
 
 
+@pytest.mark.parametrize(
+    "get_config",
+    [
+        {
+            "command_line": [
+                {
+                    "sensor": {
+                        "name": "Test",
+                        "command": 'echo \
+                            {\
+                                \\"top_level\\": {\
+                                    \\"second_level\\": {\
+                                        \\"key\\": \\"some_json_value\\",\
+                                        \\"another_key\\": \\"another_json_value\\",\
+                                        \\"key_three\\": \\"value_three\\"\
+                                    }\
+                                }\
+                            }',
+                        "json_attributes": ["key", "another_key", "key_three"],
+                        "json_attributes_path": "$.top_level.second_level",
+                    }
+                }
+            ]
+        }
+    ],
+)
+async def test_update_with_json_attrs_with_json_attrs_path(
+    hass: HomeAssistant, load_yaml_integration: None
+) -> None:
+    """Test using json_attributes_path to select a different part of the json object as root."""
+
+    entity_state = hass.states.get("sensor.test")
+    assert entity_state
+    assert entity_state.attributes["key"] == "some_json_value"
+    assert entity_state.attributes["another_key"] == "another_json_value"
+    assert entity_state.attributes["key_three"] == "value_three"
+    assert "top_level" not in entity_state.attributes
+    assert "second_level" not in entity_state.attributes
+
+
 @pytest.mark.parametrize(
     "get_config",
     [