Field AdminMessage.confirm_set_radio
var exit_simulatorField AdminMessage.exit_simulator
var get_channel_requestField AdminMessage.get_channel_request
DESCRIPTORconfirm_set_channelconfirm_set_radioexit_simulatorget_channel_requestget_channel_responseget_radio_request
+def exitSimulator(self)
+Tell a simulator node to exit (this message is ignored for other nodes)
def exitSimulator(self):
+ """
+ Tell a simulator node to exit (this message is ignored for other nodes)
+ """
+ p = admin_pb2.AdminMessage()
+ p.exit_simulator = True
+
+ return self._sendAdmin(p)
def getChannelByName(self, name)
Node
channelURLexitSimulatorgetChannelByNamegetDisabledChannelrequestConfigField MyNodeInfo.has_gps
var hw_modelvar hw_model_deprecatedField MyNodeInfo.hw_model
Field MyNodeInfo.hw_model_deprecated
var max_channelsvar next_hopField NodeInfo.next_hop
var numField NodeInfo.num
var hw_modelField User.hw_model
var idField User.id
error_countfirmware_versionhas_gpshw_modelhw_model_deprecatedmax_channelsmessage_timeout_msecmin_app_versionNodeInfoDESCRIPTORnext_hopnumpositionsnrUserDESCRIPTORhw_modelidlong_namemacaddrfrom . import portnums_pb2, remote_hardware_pb2
from pubsub import pub
+
def onGPIOreceive(packet, interface):
"""Callback for received GPIO responses
-
+
FIXME figure out how to do closures with methods in python"""
pb = remote_hardware_pb2.HardwareMessage()
pb.ParseFromString(packet["decoded"]["data"]["payload"])
print(f"Received RemoteHardware typ={pb.typ}, gpio_value={pb.gpio_value}")
+
class RemoteHardwareClient:
"""
This is the client code to control/monitor simple hardware built into the
@@ -51,8 +53,17 @@ class RemoteHardwareClient:
iface is the already open MeshInterface instance
"""
self.iface = iface
+ ch = iface.localNode.getChannelByName("gpio")
+ if not ch:
+ raise Exception(
+ "No gpio channel found, please create before using this (secured) service")
+ self.channelIndex = ch.index
- pub.subscribe(onGPIOreceive, "meshtastic.receive.data.REMOTE_HARDWARE_APP")
+ pub.subscribe(
+ onGPIOreceive, "meshtastic.receive.data.REMOTE_HARDWARE_APP")
+
+ def _sendHardware(self, nodeid, r):
+ return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck=True, channelIndex=self.channelIndex)
def writeGPIOs(self, nodeid, mask, vals):
"""
@@ -63,21 +74,21 @@ class RemoteHardwareClient:
r.typ = remote_hardware_pb2.HardwareMessage.Type.WRITE_GPIOS
r.gpio_mask = mask
r.gpio_value = vals
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
def readGPIOs(self, nodeid, mask):
"""Read the specified bits from GPIO inputs on the device"""
r = remote_hardware_pb2.HardwareMessage()
r.typ = remote_hardware_pb2.HardwareMessage.Type.READ_GPIOS
r.gpio_mask = mask
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
def watchGPIOs(self, nodeid, mask):
"""Watch the specified bits from GPIO inputs on the device for changes"""
r = remote_hardware_pb2.HardwareMessage()
r.typ = remote_hardware_pb2.HardwareMessage.Type.WATCH_GPIOS
r.gpio_mask = mask
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True) def onGPIOreceive(packet, interface):
"""Callback for received GPIO responses
-
+
FIXME figure out how to do closures with methods in python"""
pb = remote_hardware_pb2.HardwareMessage()
pb.ParseFromString(packet["decoded"]["data"]["payload"])
@@ -140,8 +151,17 @@ code for how you can connect to your own custom meshtastic services
iface is the already open MeshInterface instance
"""
self.iface = iface
+ ch = iface.localNode.getChannelByName("gpio")
+ if not ch:
+ raise Exception(
+ "No gpio channel found, please create before using this (secured) service")
+ self.channelIndex = ch.index
- pub.subscribe(onGPIOreceive, "meshtastic.receive.data.REMOTE_HARDWARE_APP")
+ pub.subscribe(
+ onGPIOreceive, "meshtastic.receive.data.REMOTE_HARDWARE_APP")
+
+ def _sendHardware(self, nodeid, r):
+ return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck=True, channelIndex=self.channelIndex)
def writeGPIOs(self, nodeid, mask, vals):
"""
@@ -152,21 +172,21 @@ code for how you can connect to your own custom meshtastic services
r.typ = remote_hardware_pb2.HardwareMessage.Type.WRITE_GPIOS
r.gpio_mask = mask
r.gpio_value = vals
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
def readGPIOs(self, nodeid, mask):
"""Read the specified bits from GPIO inputs on the device"""
r = remote_hardware_pb2.HardwareMessage()
r.typ = remote_hardware_pb2.HardwareMessage.Type.READ_GPIOS
r.gpio_mask = mask
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
def watchGPIOs(self, nodeid, mask):
"""Watch the specified bits from GPIO inputs on the device for changes"""
r = remote_hardware_pb2.HardwareMessage()
r.typ = remote_hardware_pb2.HardwareMessage.Type.WATCH_GPIOS
r.gpio_mask = mask
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
@@ -201,7 +221,7 @@ code for how you can connect to your own custom meshtastic services
r = remote_hardware_pb2.HardwareMessage()
r.typ = remote_hardware_pb2.HardwareMessage.Type.WATCH_GPIOS
r.gpio_mask = mask
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
@@ -224,7 +244,7 @@ are 1 will be changed
r.typ = remote_hardware_pb2.HardwareMessage.Type.WRITE_GPIOS
r.gpio_mask = mask
r.gpio_value = vals
- return self.iface.sendData(r, nodeid, portnums_pb2.REMOTE_HARDWARE_APP, wantAck = True)
+ return self._sendHardware(nodeid, r)
diff --git a/docs/meshtastic/test.html b/docs/meshtastic/test.html
index 0a4ddc7..9b68426 100644
--- a/docs/meshtastic/test.html
+++ b/docs/meshtastic/test.html
@@ -28,7 +28,7 @@
import logging
from . import util
-from . import SerialInterface, BROADCAST_NUM
+from . import SerialInterface, TCPInterface, BROADCAST_NUM
from pubsub import pub
import time
import sys
@@ -178,7 +178,26 @@ def testAll():
testThread()
for i in interfaces:
- i.close()
+def testSimulator()
+Assume that someone has launched meshtastic-native as a simulated node. +Talk to that node over TCP, do some operations and if they are successful +exit the process with a success code, else exit with a non zero exit code.
+Run with +python3 -c 'from meshtastic.test import testSimulator; testSimulator()'
def testSimulator():
+ """
+ Assume that someone has launched meshtastic-native as a simulated node.
+ Talk to that node over TCP, do some operations and if they are successful
+ exit the process with a success code, else exit with a non zero exit code.
+
+ Run with
+ python3 -c 'from meshtastic.test import testSimulator; testSimulator()'
+ """
+ logging.basicConfig(level=logging.DEBUG if False else logging.INFO)
+ logging.info("Connecting to simulator on localhost!")
+ iface = TCPInterface("localhost")
+ iface.showInfo()
+ iface.localNode.showInfo()
+ iface.localNode.exitSimulator()
+ iface.close()
+ logging.info("Integration test successful!")
+ sys.exit(0)
def testThread(numTests=50)
subscribetestAlltestSendtestSimulatortestThread