Add CLI Console: run companion commands with visible output from the dashboard

custom_components/meshcore/__init__.py

@@ -1,71 +1,71 @@
 """The MeshCore integration."""
 from __future__ import annotations
 
 import asyncio
 import json
 import logging
 import time
 from pathlib import Path
 from datetime import timedelta
 from meshcore.events import EventType
 
 from .const import (
     CONF_REPEATER_TELEMETRY_ENABLED,
     CONF_TRACKED_CLIENTS,
 )
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.components.http import StaticPathConfig
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers import issue_registry as ir
 from homeassistant.helpers.device_registry import DeviceEntry
 
 
 from .const import (
     DOMAIN,
     CONF_CONNECTION_TYPE,
     CONF_NAME,
     CONF_USB_PATH,
     CONF_BLE_ADDRESS,
     CONF_TCP_HOST,
     CONF_TCP_PORT,
     CONF_BAUDRATE,
     CONF_PUBKEY,
     CONF_REPEATER_SUBSCRIPTIONS,
     CONF_LIMIT_DISCOVERED_CONTACTS,
     CONF_MAX_DISCOVERED_CONTACTS,
     DEFAULT_MAX_DISCOVERED_CONTACTS,
     CONF_FLOOD_SCOPES,
     CONF_MESSAGES_INTERVAL,
     DEFAULT_UPDATE_TICK,
     REPAIR_PUBKEY_CHANGED,
     CONF_CONTACT_DISCOVERY_MODE,
     MODE_FULL,
     MODE_DATA_ONLY,
     MODE_OFF,
     get_contact_discovery_mode,
 )
 from .coordinator import MeshCoreDataUpdateCoordinator
 from .meshcore_api import MeshCoreAPI
 from .map_uploader import MeshCoreMapUploader
 from .mqtt_uploader import MeshCoreMqttUploader
 from .services import async_setup_services, async_unload_services
 from .utils import (
     create_message_correlation_key,
     load_flood_scope_keys,
     match_flood_scope,
     parse_and_decrypt_rx_log,
     parse_rx_log_data,
     sanitize_event_data,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
 # List of platforms to set up
-PLATFORMS = [Platform.SENSOR, Platform.BINARY_SENSOR, Platform.SELECT, Platform.TEXT, Platform.DEVICE_TRACKER]
+PLATFORMS = [Platform.SENSOR, Platform.BINARY_SENSOR, Platform.SELECT, Platform.TEXT, Platform.DEVICE_TRACKER, Platform.BUTTON]
 STATIC_PATH_REGISTERED_KEY = f"{DOMAIN}_static_path_registered"
 
 
@@ -483,7 +483,7 @@
     if connected and api.mesh_core:
         try:
             _LOGGER.info("Loading contacts from device on initialization...")
             contacts_changed = await api.mesh_core.ensure_contacts(follow=False)
 
             # Index contacts by 12-char prefix
             coordinator._contacts = {}

custom_components/meshcore/button.py

@@ -0,0 +1,104 @@
+"""Button platform for MeshCore integration.
+
+Provides the CLI Console controls as button entities so they appear on the
+device page automatically and render compactly (unlike a full button *card*).
+Created only when CONF_CLI_CONSOLE_ENABLED is set.
+"""
+from __future__ import annotations
+
+import logging
+
+from homeassistant.components.button import ButtonEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import (
+    CONF_CLI_CONSOLE_ENABLED,
+    DOMAIN,
+    ENTITY_DOMAIN_BUTTON,
+    SERVICE_EXECUTE_COMMAND_UI,
+)
+from .utils import format_entity_id
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up MeshCore button entities from a config entry."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+
+    entities: list[ButtonEntity] = []
+    if entry.data.get(CONF_CLI_CONSOLE_ENABLED, False):
+        entities.append(MeshCoreCLIRunButton(coordinator))
+        entities.append(MeshCoreCLIClearButton(coordinator))
+
+    if entities:
+        async_add_entities(entities)
+
+
+class _MeshCoreCLIButton(CoordinatorEntity, ButtonEntity):
+    """Shared base for CLI Console buttons.
+
+    Deliberately NOT attached to the companion device and hidden by default:
+    the console only works as a dashboard card (the device page can't render the
+    transcript), so surfacing these on the device page would be misleading.
+    They're used by referencing their entity_id in the CLI Console card.
+    """
+
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+    _attr_entity_registry_visible_default = False
+
+    def __init__(self, coordinator) -> None:
+        super().__init__(coordinator)
+        self.coordinator = coordinator
+
+
+class MeshCoreCLIRunButton(_MeshCoreCLIButton):
+    """Runs the command in text.meshcore_command through the CLI Console."""
+
+    _attr_name = "CLI Run Command"
+    _attr_icon = "mdi:play"
+
+    def __init__(self, coordinator) -> None:
+        super().__init__(coordinator)
+        public_key_short = coordinator.pubkey[:6] if coordinator.pubkey else ""
+        self._attr_unique_id = f"{coordinator.config_entry.entry_id}_cli_run"
+        self.entity_id = format_entity_id(
+            ENTITY_DOMAIN_BUTTON, public_key_short, "cli_run"
+        )
+
+    async def async_press(self) -> None:
+        """Execute the command in the input helper and record its output."""
+        await self.hass.services.async_call(
+            DOMAIN,
+            SERVICE_EXECUTE_COMMAND_UI,
+            {
+                "entry_id": self.coordinator.config_entry.entry_id,
+                "record_to_console": True,
+            },
+            blocking=True,
+        )
+
+
+class MeshCoreCLIClearButton(_MeshCoreCLIButton):
+    """Clears the CLI Console transcript."""
+
+    _attr_name = "CLI Clear Console"
+    _attr_icon = "mdi:notification-clear-all"
+
+    def __init__(self, coordinator) -> None:
+        super().__init__(coordinator)
+        public_key_short = coordinator.pubkey[:6] if coordinator.pubkey else ""
+        self._attr_unique_id = f"{coordinator.config_entry.entry_id}_cli_clear"
+        self.entity_id = format_entity_id(
+            ENTITY_DOMAIN_BUTTON, public_key_short, "cli_clear"
+        )
+
+    async def async_press(self) -> None:
+        """Empty the console transcript."""
+        self.coordinator.clear_cli_console()

custom_components/meshcore/config_flow.py

@@ -61,6 +61,7 @@
     DEFAULT_SELF_TELEMETRY_INTERVAL,
     CONF_SELF_DIAGNOSTICS_ENABLED,
     CONF_SELF_DIAGNOSTICS_INTERVAL,
+    CONF_CLI_CONSOLE_ENABLED,
     DEFAULT_SELF_DIAGNOSTICS_INTERVAL,
     CONF_MAP_UPLOAD_ENABLED,
     CONF_AUTO_CLEANUP_STALE_CONTACTS,
@@ -930,6 +931,7 @@ async def async_step_global_settings(self, user_input=None):
             new_data[CONF_SELF_TELEMETRY_INTERVAL] = user_input[CONF_SELF_TELEMETRY_INTERVAL]
             new_data[CONF_SELF_DIAGNOSTICS_ENABLED] = user_input[CONF_SELF_DIAGNOSTICS_ENABLED]
             new_data[CONF_SELF_DIAGNOSTICS_INTERVAL] = user_input[CONF_SELF_DIAGNOSTICS_INTERVAL]
+            new_data[CONF_CLI_CONSOLE_ENABLED] = user_input[CONF_CLI_CONSOLE_ENABLED]
             new_data[CONF_MAP_UPLOAD_ENABLED] = user_input[CONF_MAP_UPLOAD_ENABLED]
             new_data[CONF_AUTO_CLEANUP_STALE_CONTACTS] = user_input[CONF_AUTO_CLEANUP_STALE_CONTACTS]
             new_data[CONF_STALE_CONTACT_DAYS] = user_input[CONF_STALE_CONTACT_DAYS]
@@ -955,6 +957,7 @@ async def async_step_global_settings(self, user_input=None):
         current_telemetry_interval = self.config_entry.data.get(CONF_SELF_TELEMETRY_INTERVAL, DEFAULT_SELF_TELEMETRY_INTERVAL)
         current_diagnostics_enabled = self.config_entry.data.get(CONF_SELF_DIAGNOSTICS_ENABLED, False)
         current_diagnostics_interval = self.config_entry.data.get(CONF_SELF_DIAGNOSTICS_INTERVAL, DEFAULT_SELF_DIAGNOSTICS_INTERVAL)
+        current_cli_console_enabled = self.config_entry.data.get(CONF_CLI_CONSOLE_ENABLED, False)
         current_map_upload_enabled = self.config_entry.data.get(CONF_MAP_UPLOAD_ENABLED, False)
         current_auto_cleanup = self.config_entry.data.get(CONF_AUTO_CLEANUP_STALE_CONTACTS, False)
         current_stale_days = self.config_entry.data.get(CONF_STALE_CONTACT_DAYS, DEFAULT_STALE_CONTACT_DAYS)
@@ -973,6 +976,7 @@ async def async_step_global_settings(self, user_input=None):
                 vol.Optional(CONF_SELF_TELEMETRY_INTERVAL, default=current_telemetry_interval): vol.All(cv.positive_int, vol.Range(min=60, max=3600)),
                 vol.Optional(CONF_SELF_DIAGNOSTICS_ENABLED, default=current_diagnostics_enabled): cv.boolean,
                 vol.Optional(CONF_SELF_DIAGNOSTICS_INTERVAL, default=current_diagnostics_interval): vol.All(cv.positive_int, vol.Range(min=60, max=3600)),
+                vol.Optional(CONF_CLI_CONSOLE_ENABLED, default=current_cli_console_enabled): cv.boolean,
                 vol.Optional(CONF_MAP_UPLOAD_ENABLED, default=current_map_upload_enabled): cv.boolean,
                 vol.Optional(CONF_AUTO_CLEANUP_STALE_CONTACTS, default=current_auto_cleanup): cv.boolean,
                 vol.Optional(CONF_STALE_CONTACT_DAYS, default=current_stale_days): vol.All(cv.positive_int, vol.Range(min=1, max=365)),

custom_components/meshcore/const.py

@@ -31,6 +31,7 @@
 SERVICE_SEND_CHANNEL_MESSAGE: Final = "send_channel_message"
 SERVICE_EXECUTE_COMMAND: Final = "execute_command"
 SERVICE_EXECUTE_COMMAND_UI: Final = "execute_command_ui"
+SERVICE_CLI_CLEAR: Final = "cli_console_clear"
 SERVICE_MESSAGE_SCRIPT: Final = "send_ui_message"
 SERVICE_ADD_SELECTED_CONTACT: Final = "add_selected_contact"
 SERVICE_REMOVE_SELECTED_CONTACT: Final = "remove_selected_contact"
@@ -53,13 +54,17 @@
 ATTR_COMMAND: Final = "command"
 ATTR_ENTRY_ID: Final = "entry_id"
 ATTR_SCOPE: Final = "scope"
+# When set on execute_command / execute_command_ui, the command/response pair is
+# recorded to the CLI Console transcript and the meshcore_cli_response event fires.
+ATTR_RECORD_TO_CONSOLE: Final = "record_to_console"
 
 # Platform constants
 PLATFORM_MESSAGE: Final = "message"
 
 # Entity naming constants
 ENTITY_DOMAIN_BINARY_SENSOR: Final = "binary_sensor"
 ENTITY_DOMAIN_SENSOR: Final = "sensor"
+ENTITY_DOMAIN_BUTTON: Final = "button"
 DEFAULT_DEVICE_NAME: Final = "meshcore"
 MESSAGES_SUFFIX: Final = "messages"
 CONTACT_SUFFIX: Final = "contact"
@@ -127,6 +132,17 @@ def get_contact_discovery_mode(config_entry) -> str:
 CONF_SELF_DIAGNOSTICS_INTERVAL: Final = "self_diagnostics_interval"
 DEFAULT_SELF_DIAGNOSTICS_INTERVAL: Final = 300  # 5 minutes in seconds
 
+# CLI console settings. An interactive command surface for the local companion
+# radio: execute_command / execute_command_ui called with record_to_console: true
+# record the command and its response into a sensor transcript so output is
+# visible in the UI (a plain execute_command_ui discards the response). It does
+# NOT stream LOG_DATA / RX_LOG packet noise — only command/response pairs.
+CONF_CLI_CONSOLE_ENABLED: Final = "cli_console_enabled"
+# Number of command/response pairs kept in the rolling console transcript.
+CLI_CONSOLE_MAX_LINES: Final = 50
+# Event fired after a CLI console command completes (for logbook/automations).
+EVENT_CLI_RESPONSE: Final = f"{DOMAIN}_cli_response"
+
 # STATS_CORE `errors` is a bitmask of radio dispatcher fault events, not a
 # count. Each bit latches on first occurrence and clears only on a radio
 # reboot. Bit values mirror the firmware (_err_flags in MeshCore Dispatcher.h);

custom_components/meshcore/coordinator.py

@@ -5,6 +5,7 @@
 import logging
 import random
 import time
+from collections import deque
 from datetime import timedelta
 from typing import Any, Dict
 
@@ -40,6 +41,7 @@
     CONF_SELF_TELEMETRY_ENABLED,
     CONF_SELF_TELEMETRY_INTERVAL,
     DEFAULT_SELF_TELEMETRY_INTERVAL,
+    CLI_CONSOLE_MAX_LINES,
     CONF_SELF_DIAGNOSTICS_ENABLED,
     CONF_SELF_DIAGNOSTICS_INTERVAL,
     DEFAULT_SELF_DIAGNOSTICS_INTERVAL,
@@ -127,6 +129,16 @@ def __init__(
         # Get name and pubkey from config_entry.data (not options)
         self.name = config_entry.data.get(CONF_NAME)
         self.pubkey = config_entry.data.get(CONF_PUBKEY)
+
+        # Rolling transcript for the CLI console sensor: each entry is a dict of
+        # {timestamp, command, response, is_error}. Bounded so the sensor's
+        # attribute payload stays small regardless of how much the console is
+        # used. The sensor (when CONF_CLI_CONSOLE_ENABLED) registers itself here
+        # so record_cli_console() can push fresh state immediately.
+        self.cli_console_history: deque[dict[str, Any]] = deque(
+            maxlen=CLI_CONSOLE_MAX_LINES
+        )
+        self.cli_console_sensor: Any = None
 
         # Set up device info that entities can reference
         self._firmware_version = None
@@ -256,6 +268,39 @@ def __init__(
         # Set of pubkey prefixes that have been updated and need sensor refresh
         self._dirty_contacts = set()
 
+    def record_cli_console(
+        self, command: str, response: Any, is_error: bool = False
+    ) -> None:
+        """Append a command/response pair to the CLI console transcript.
+
+        Pushes fresh state to the console sensor immediately when one is
+        registered (CONF_CLI_CONSOLE_ENABLED). No-ops gracefully when the
+        console is disabled, so the execute_command record_to_console path can
+        call this unconditionally.
+        """
+        self.cli_console_history.append({
+            "timestamp": int(time.time()),
+            "command": command,
+            "response": response,
+            "is_error": bool(is_error),
+        })
+        sensor = self.cli_console_sensor
+        if sensor is not None:
+            try:
+                sensor.async_write_ha_state()
+            except Exception as ex:  # pragma: no cover - defensive
+                _LOGGER.debug("Failed to update CLI console sensor: %s", ex)
+
+    def clear_cli_console(self) -> None:
+        """Empty the CLI console transcript and refresh the sensor."""
+        self.cli_console_history.clear()
+        sensor = self.cli_console_sensor
+        if sensor is not None:
+            try:
+                sensor.async_write_ha_state()
+            except Exception as ex:  # pragma: no cover - defensive
+                _LOGGER.debug("Failed to update CLI console sensor: %s", ex)
+
     def mark_contact_dirty(self, pubkey_prefix: str):
         """Mark a contact as needing update (for performance optimization).