send raw packets

README.md

@@ -518,6 +518,8 @@ All commands are async methods that return `Event` objects. Commands are organiz
 | **Device Actions** ||||
 | `send_advert(flood=False)` | `flood: bool` | `OK` | Send advertisement (optionally flood network) |
 | `reboot()` | None | None | Reboot device (no response expected) |
+| `run_cli_command()` | `cmd: str` | `CLI_REPLY/ERROR` | Send a cli command to device |
+| `send_raw_packet()` | `data: str/bytes, priority=0` | `OK/ERROR` | Send a raw packet through device, can be an hex string or bytes |
 | **Security** ||||
 | `export_private_key()` | None | `PRIVATE_KEY/DISABLED` | Export device private key (requires PIN auth & enabled firmware) |
 | `import_private_key(key)` | `key: bytes` | `OK` | Import private key to device |

src/meshcore/commands/device.py

@@ -16,15 +16,26 @@ class DeviceCommands(CommandHandlerBase):
         b1 = bytearray(b"\x01\x03      mccli")
         return await self.send(b1, [EventType.SELF_INFO, 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])
+
     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:
+    async def send_raw_packet(self, data, priority=0):
-        logger.debug("Sending device query command")
+        if isinstance(data, str): # got a string (assuming hex), change it to bytes
-        return await self.send(b"\x16\x03", [EventType.DEVICE_INFO, EventType.ERROR])
+            data=bytes.fromhex(data)
+
+        cmd_data = bytearray([CommandType.SEND_RAW_PACKET.value])
+        cmd_data.append(priority)
+        cmd_data.extend(data)
+
+        logger.debug(f"Sending raw packet {data.hex()}")
+        return await self.send(cmd_data, [EventType.OK, EventType.ERROR])
 
     async def send_advert(self, flood: bool = False) -> Event:
         logger.debug(f"Sending advertisement command (flood={flood})")

src/meshcore/commands/messaging.py

@@ -72,7 +72,6 @@ class MessagingCommands(CommandHandlerBase):
             timestamp = int(time.time())
 
         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.")