support for run_cli_command

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "meshcore"
-version = "2.3.8"
+version = "2.3.9"
 authors = [
   { name="Florent de Lamotte", email="florent@frizoncorrea.fr" },
   { name="Alex Wolden", email="awolden@gmail.com" },

src/meshcore/commands/device.py

@@ -16,6 +16,12 @@ class DeviceCommands(CommandHandlerBase):
         b1 = bytearray(b"\x01\x03      mccli")
         return await self.send(b1, [EventType.SELF_INFO, EventType.ERROR])
 
+    async def run_cli_command(self, cmd: str) -> Event:
+        logger.debug(f"Sending cli command: {cmd}")
+        cmd_data = bytearray([CommandType.RUN_CLI_COMMAND.value])
+        cmd_data.extend(cmd.encode("utf-8"))
+        return await self.send(cmd_data, [EventType.CLI_REPLY, EventType.ERROR])
+
     async def send_device_query(self) -> Event:
         logger.debug("Sending device query command")
         return await self.send(b"\x16\x03", [EventType.DEVICE_INFO, EventType.ERROR])

src/meshcore/commands/messaging.py

@@ -4,7 +4,7 @@ from typing import Optional, Union
 from hashlib import sha256
 
 from ..events import Event, EventType
-from ..packets import CommandType
+from ..packets import CommandType, TxtType, AdvType
 from .base import CommandHandlerBase, DestinationType, _validate_destination
 
 logger = logging.getLogger("meshcore")
@@ -61,23 +61,33 @@ class MessagingCommands(CommandHandlerBase):
         return await self.send(data, [EventType.MSG_SENT, EventType.ERROR])
 
     async def send_cmd(
-        self, dst: DestinationType, cmd: str, timestamp: Optional[int] = None
+        self, dst: DestinationType, cmd: str, timestamp: Optional[int] = None,
+        dst_type = None # if None, will have to get a contact to know this
     ) -> Event:
         dst_bytes = _validate_destination(dst)
         logger.debug(f"Sending command to {dst_bytes.hex()}: {cmd}")
 
         if timestamp is None:
             import time
-
             timestamp = int(time.time())
 
-        data = (
-            b"\x02\x01\x00"
-            + timestamp.to_bytes(4, "little")
-            + dst_bytes
-            + cmd.encode("utf-8")
-        )
-        return await self.send(data, [EventType.MSG_SENT, EventType.ERROR])
+        if dst_type is None:
+            if isinstance(dst, dict) and "type" in dst :
+                dst_type = dst["type"]
+            else: # assume destination is a repeater
+                logger.warning("Can't determine destination type, please ensure contacts first or specify dst_type when calling `send_cmd`. Assuming it's a repeater.")
+                dst_type = AdvType.REPEATER.value
+
+        cmd_data = bytearray([CommandType.SEND_TXT_MSG.value])
+        if dst_type == AdvType.REPEATER.value :
+            cmd_data.append(TxtType.CLI_DATA.value)
+        else:
+            cmd_data.append(TxtType.CLI_CMD.value)
+        cmd_data.append(0) # first and only attempt
+        cmd_data.extend(timestamp.to_bytes(4, "little"))
+        cmd_data.extend(dst_bytes)
+        cmd_data.extend(cmd.encode("utf-8"))
+        return await self.send(cmd_data, [EventType.MSG_SENT, EventType.ERROR])
 
     async def send_msg(
         self, dst: DestinationType, msg: str, timestamp: Optional[int] = None,

src/meshcore/events.py

@@ -25,6 +25,7 @@ class EventType(Enum):
     NEXT_CONTACT = "next_contact"
     AUTOADD_CONFIG = "autoadd_config"
     ADVERT_PATH = "advert_path"
+    CLI_REPLY = "cli_reply"
 
     # Push notifications
     ADVERTISEMENT = "advertisement"

src/meshcore/packets.py

@@ -1,5 +1,18 @@
 from enum import Enum
 
+class AdvType(Enum):
+    NONE = 0x00
+    CHAT = 0x01
+    REPEATER = 0x02
+    ROOM = 0x03
+    SENSOR = 0x04
+
+class TxtType(Enum):
+    PLAIN = 0x00
+    CLI_DATA = 0x01
+    SIGNED_PLAIN = 0x02
+    CLI_CMD = 0x03
+
 class AnonReqType(Enum):
     REGIONS = 0x01
     OWNER = 0x02
@@ -75,6 +88,8 @@ class CommandType(Enum):
     SET_PATH_HASH_MODE = 61
     SET_DEFAULT_FLOOD_SCOPE = 63
     GET_DEFAULT_FLOOD_SCOPE = 64
+    SEND_RAW_PACKET = 65
+    RUN_CLI_COMMAND = 66
 
 # Packet prefixes for the protocol
 class PacketType(Enum):
@@ -107,6 +122,7 @@ class PacketType(Enum):
     ALLOWED_REPEAT_FREQ = 26
     CHANNEL_DATA_RECV = 27
     DEFAULT_FLOOD_SCOPE = 28
+    CLI_REPLY = 29
 
     # Push notifications
     ADVERTISEMENT = 0x80

src/meshcore/reader.py

@@ -1052,6 +1052,11 @@ class MessageReader:
                 res = {"rx_delay": rx_delay, "airtime_factor": airtime_factor}
                 await self.dispatcher.dispatch(Event(EventType.TUNING_PARAMS, res))
 
+            elif packet_type_value == PacketType.CLI_REPLY.value:
+                res = {}
+                res["text"] = data[1:].decode("utf-8", "ignore")
+                await self.dispatcher.dispatch(Event(EventType.CLI_REPLY, res))
+
             else:
                 logger.debug(f"Unhandled data received {data}")
                 logger.debug(f"Unhandled packet type: {packet_type_value}")