Module meshtastic.__main__
Main Meshtastic
Expand source code
#!python3
""" Main Meshtastic
"""
import argparse
import platform
import logging
import sys
import time
import yaml
from pubsub import pub
import pyqrcode
import pkg_resources
import meshtastic.util
import meshtastic.test
from . import remote_hardware
from . import portnums_pb2, channel_pb2, radioconfig_pb2
from .globals import Globals
have_tunnel = platform.system() == 'Linux'
"""We only import the tunnel code if we are on a platform that can run it. """
def onReceive(packet, interface):
"""Callback invoked when a packet arrives"""
our_globals = Globals.getInstance()
args = our_globals.get_args()
try:
d = packet.get('decoded')
logging.debug(f'in onReceive() d:{d}')
# Exit once we receive a reply
if args and args.sendtext and packet["to"] == interface.myInfo.my_node_num and d["portnum"] == portnums_pb2.PortNum.TEXT_MESSAGE_APP:
interface.close() # after running command then exit
# Reply to every received message with some stats
if args and args.reply:
msg = d.get('text')
if msg:
rxSnr = packet['rxSnr']
hopLimit = packet['hopLimit']
print(f"message: {msg}")
reply = "got msg \'{}\' with rxSnr: {} and hopLimit: {}".format(msg, rxSnr, hopLimit)
print("Sending reply: ", reply)
interface.sendText(reply)
except Exception as ex:
print(ex)
def onConnection(interface, topic=pub.AUTO_TOPIC):
"""Callback invoked when we connect/disconnect from a radio"""
print(f"Connection changed: {topic.getName()}")
def getPref(attributes, name):
"""Get a channel or preferences value"""
objDesc = attributes.DESCRIPTOR
field = objDesc.fields_by_name.get(name)
if not field:
print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not get it.")
print(f"Choices in sorted order are:")
names = []
for f in objDesc.fields:
names.append(f'{f.name}')
for temp_name in sorted(names):
print(f" {temp_name}")
return
# okay - try to read the value
try:
try:
val = getattr(attributes, name)
except TypeError:
# The getter didn't like our arg type guess try again as a string
val = getattr(attributes, name)
# succeeded!
print(f"{name}: {str(val)}")
except Exception as ex:
print(f"Can't get {name} due to {ex}")
def setPref(attributes, name, valStr):
"""Set a channel or preferences value"""
objDesc = attributes.DESCRIPTOR
field = objDesc.fields_by_name.get(name)
if not field:
print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not set it.")
print(f"Choices in sorted order are:")
names = []
for f in objDesc.fields:
names.append(f'{f.name}')
for temp_name in sorted(names):
print(f" {temp_name}")
return
val = meshtastic.util.fromStr(valStr)
enumType = field.enum_type
# pylint: disable=C0123
if enumType and type(val) == str:
# We've failed so far to convert this string into an enum, try to find it by reflection
e = enumType.values_by_name.get(val)
if e:
val = e.number
else:
print(f"{name} does not have an enum called {val}, so you can not set it.")
print(f"Choices in sorted order are:")
names = []
for f in enumType.values:
names.append(f'{f.name}')
for temp_name in sorted(names):
print(f" {temp_name}")
return
# okay - try to read the value
try:
try:
setattr(attributes, name, val)
except TypeError:
# The setter didn't like our arg type guess try again as a string
setattr(attributes, name, valStr)
# succeeded!
print(f"Set {name} to {valStr}")
except Exception as ex:
print(f"Can't set {name} due to {ex}")
def onConnected(interface):
"""Callback invoked when we connect to a radio"""
closeNow = False # Should we drop the connection after we finish?
try:
our_globals = Globals.getInstance()
args = our_globals.get_args()
print("Connected to radio")
def getNode():
"""This operation could be expensive, so we try to cache the results"""
targetNode = our_globals.get_target_node()
if not targetNode:
targetNode = interface.getNode(args.destOrLocal)
our_globals.set_target_node(targetNode)
return targetNode
if args.setlat or args.setlon or args.setalt:
closeNow = True
alt = 0
lat = 0.0
lon = 0.0
prefs = interface.localNode.radioConfig.preferences
if args.setalt:
alt = int(args.setalt)
prefs.fixed_position = True
print(f"Fixing altitude at {alt} meters")
if args.setlat:
lat = float(args.setlat)
prefs.fixed_position = True
print(f"Fixing latitude at {lat} degrees")
if args.setlon:
lon = float(args.setlon)
prefs.fixed_position = True
print(f"Fixing longitude at {lon} degrees")
print("Setting device position")
# can include lat/long/alt etc: latitude = 37.5, longitude = -122.1
interface.sendPosition(lat, lon, alt)
interface.localNode.writeConfig()
elif not args.no_time:
# We normally provide a current time to the mesh when we connect
interface.sendPosition()
if args.set_owner:
closeNow = True
print(f"Setting device owner to {args.set_owner}")
getNode().setOwner(args.set_owner)
if args.pos_fields:
# If --pos-fields invoked with args, set position fields
closeNow = True
prefs = getNode().radioConfig.preferences
allFields = 0
try:
for field in args.pos_fields:
v_field = radioconfig_pb2.PositionFlags.Value(field)
allFields |= v_field
except ValueError:
print("ERROR: supported position fields are:")
print(radioconfig_pb2.PositionFlags.keys())
print("If no fields are specified, will read and display current value.")
else:
print(f"Setting position fields to {allFields}")
setPref(prefs, 'position_flags', ('%d' % allFields))
print("Writing modified preferences to device")
getNode().writeConfig()
elif args.pos_fields is not None:
# If --pos-fields invoked without args, read and display current value
closeNow = True
prefs = getNode().radioConfig.preferences
fieldNames = []
for bit in radioconfig_pb2.PositionFlags.values():
if prefs.position_flags & bit:
fieldNames.append(radioconfig_pb2.PositionFlags.Name(bit))
print(' '.join(fieldNames))
if args.set_team:
closeNow = True
try:
v_team = meshtastic.mesh_pb2.Team.Value(args.set_team.upper())
except ValueError:
v_team = 0
print(f"ERROR: Team \'{args.set_team}\' not found.")
print("Try a team name from the sorted list below, or use 'CLEAR' for unaffiliated:")
print(sorted(meshtastic.mesh_pb2.Team.keys()))
else:
print(f"Setting team to {meshtastic.mesh_pb2.Team.Name(v_team)}")
getNode().setOwner(team=v_team)
if args.set_ham:
closeNow = True
print(f"Setting Ham ID to {args.set_ham} and turning off encryption")
getNode().setOwner(args.set_ham, is_licensed=True)
# Must turn off encryption on primary channel
getNode().turnOffEncryptionOnPrimaryChannel()
if args.reboot:
closeNow = True
getNode().reboot()
if args.sendtext:
closeNow = True
channelIndex = 0
if args.ch_index is not None:
channelIndex = int(args.ch_index)
ch = getNode().getChannelByChannelIndex(channelIndex)
if ch and ch.role != channel_pb2.Channel.Role.DISABLED:
print(f"Sending text message {args.sendtext} to {args.destOrAll} on channelIndex:{channelIndex}")
interface.sendText(args.sendtext, args.destOrAll, wantAck=True, channelIndex=channelIndex)
else:
meshtastic.util.our_exit(f"Warning: {channelIndex} is not a valid channel. Channel must not be DISABLED.")
if args.sendping:
payload = str.encode("test string")
print(f"Sending ping message to {args.destOrAll}")
interface.sendData(payload, args.destOrAll, portNum=portnums_pb2.PortNum.REPLY_APP,
wantAck=True, wantResponse=True)
if args.gpio_wrb or args.gpio_rd or args.gpio_watch:
rhc = remote_hardware.RemoteHardwareClient(interface)
if args.gpio_wrb:
bitmask = 0
bitval = 0
for wrpair in (args.gpio_wrb or []):
bitmask |= 1 << int(wrpair[0])
bitval |= int(wrpair[1]) << int(wrpair[0])
print(f"Writing GPIO mask 0x{bitmask:x} with value 0x{bitval:x} to {args.dest}")
rhc.writeGPIOs(args.dest, bitmask, bitval)
closeNow = True
if args.gpio_rd:
bitmask = int(args.gpio_rd, 16)
print(f"Reading GPIO mask 0x{bitmask:x} from {args.dest}")
interface.mask = bitmask
rhc.readGPIOs(args.dest, bitmask, None)
if not interface.noProto:
# wait up to X seconds for a response
for _ in range(10):
time.sleep(1)
if interface.gotResponse:
break
logging.debug(f'end of gpio_rd')
if args.gpio_watch:
bitmask = int(args.gpio_watch, 16)
print(f"Watching GPIO mask 0x{bitmask:x} from {args.dest}. Press ctrl-c to exit")
while True:
rhc.watchGPIOs(args.dest, bitmask)
time.sleep(1)
# handle settings
if args.set:
closeNow = True
prefs = getNode().radioConfig.preferences
# Handle the int/float/bool arguments
for pref in args.set:
setPref(prefs, pref[0], pref[1])
print("Writing modified preferences to device")
getNode().writeConfig()
if args.configure:
with open(args.configure[0], encoding='utf8') as file:
configuration = yaml.safe_load(file)
closeNow = True
if 'owner' in configuration:
print(f"Setting device owner to {configuration['owner']}")
getNode().setOwner(configuration['owner'])
if 'channel_url' in configuration:
print("Setting channel url to", configuration['channel_url'])
getNode().setURL(configuration['channel_url'])
if 'location' in configuration:
alt = 0
lat = 0.0
lon = 0.0
prefs = interface.localNode.radioConfig.preferences
if 'alt' in configuration['location']:
alt = int(configuration['location']['alt'])
prefs.fixed_position = True
print(f"Fixing altitude at {alt} meters")
if 'lat' in configuration['location']:
lat = float(configuration['location']['lat'])
prefs.fixed_position = True
print(f"Fixing latitude at {lat} degrees")
if 'lon' in configuration['location']:
lon = float(configuration['location']['lon'])
prefs.fixed_position = True
print(f"Fixing longitude at {lon} degrees")
print("Setting device position")
interface.sendPosition(lat, lon, alt)
interface.localNode.writeConfig()
if 'user_prefs' in configuration:
prefs = getNode().radioConfig.preferences
for pref in configuration['user_prefs']:
setPref(prefs, pref, str(configuration['user_prefs'][pref]))
print("Writing modified preferences to device")
getNode().writeConfig()
if args.export_config:
# export the configuration (the opposite of '--configure')
closeNow = True
export_config(interface)
if args.seturl:
closeNow = True
getNode().setURL(args.seturl)
# handle changing channels
if args.ch_add:
closeNow = True
if len(args.ch_add) > 10:
meshtastic.util.our_exit("Warning: Channel name must be shorter. Channel not added.")
n = getNode()
ch = n.getChannelByName(args.ch_add)
if ch:
meshtastic.util.our_exit(f"Warning: This node already has a '{args.ch_add}' channel. No changes were made.")
else:
# get the first channel that is disabled (i.e., available)
ch = n.getDisabledChannel()
if not ch:
meshtastic.util.our_exit("Warning: No free channels were found")
chs = channel_pb2.ChannelSettings()
chs.psk = meshtastic.util.genPSK256()
chs.name = args.ch_add
ch.settings.CopyFrom(chs)
ch.role = channel_pb2.Channel.Role.SECONDARY
print(f"Writing modified channels to device")
n.writeChannel(ch.index)
if args.ch_del:
closeNow = True
channelIndex = our_globals.get_channel_index()
if channelIndex is None:
meshtastic.util.our_exit("Warning: Need to specify '--ch-index' for '--ch-del'.", 1)
else:
if channelIndex == 0:
meshtastic.util.our_exit("Warning: Cannot delete primary channel.", 1)
else:
print(f"Deleting channel {channelIndex}")
ch = getNode().deleteChannel(channelIndex)
ch_changes = [args.ch_longslow, args.ch_longfast,
args.ch_mediumslow, args.ch_mediumfast,
args.ch_shortslow, args.ch_shortfast]
any_primary_channel_changes = any(x for x in ch_changes)
if args.ch_set or any_primary_channel_changes or args.ch_enable or args.ch_disable:
closeNow = True
channelIndex = our_globals.get_channel_index()
if channelIndex is None:
if any_primary_channel_changes:
# we assume that they want the primary channel if they're setting range values
channelIndex = 0
else:
meshtastic.util.our_exit("Warning: Need to specify '--ch-index'.", 1)
ch = getNode().channels[channelIndex]
if any_primary_channel_changes or args.ch_enable or args.ch_disable:
if channelIndex == 0 and not any_primary_channel_changes:
meshtastic.util.our_exit("Warning: Cannot enable/disable PRIMARY channel.")
if channelIndex != 0:
if any_primary_channel_changes:
meshtastic.util.our_exit("Warning: Standard channel settings can only be applied to the PRIMARY channel")
enable = True # default to enable
if args.ch_enable:
enable = True
if args.ch_disable:
enable = False
def setSimpleChannel(modem_config):
"""Set one of the simple modem_config only based channels"""
# Completely new channel settings
chs = channel_pb2.ChannelSettings()
chs.modem_config = modem_config
chs.psk = bytes([1]) # Use default channel psk 1
ch.settings.CopyFrom(chs)
# handle the simple channel set commands
if args.ch_longslow:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr48Sf4096)
if args.ch_longfast:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw31_25Cr48Sf512)
if args.ch_mediumslow:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr46Sf2048)
if args.ch_mediumfast:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr47Sf1024)
if args.ch_shortslow:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr45Sf128)
if args.ch_shortfast:
setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw500Cr45Sf128)
# Handle the channel settings
for pref in (args.ch_set or []):
if pref[0] == "psk":
ch.settings.psk = meshtastic.util.fromPSK(pref[1])
else:
setPref(ch.settings, pref[0], pref[1])
enable = True # If we set any pref, assume the user wants to enable the channel
if enable:
ch.role = channel_pb2.Channel.Role.PRIMARY if (
channelIndex == 0) else channel_pb2.Channel.Role.SECONDARY
else:
ch.role = channel_pb2.Channel.Role.DISABLED
print(f"Writing modified channels to device")
getNode().writeChannel(channelIndex)
if args.info:
print("")
if not args.dest: # If we aren't trying to talk to our local node, don't show it
interface.showInfo()
print("")
getNode().showInfo()
closeNow = True # FIXME, for now we leave the link up while talking to remote nodes
print("")
if args.get:
closeNow = True
prefs = getNode().radioConfig.preferences
# Handle the int/float/bool arguments
for pref in args.get:
getPref(prefs, pref[0])
print("Completed getting preferences")
if args.nodes:
closeNow = True
interface.showNodes()
if args.qr:
closeNow = True
url = interface.localNode.getURL(includeAll=False)
print(f"Primary channel URL {url}")
qr = pyqrcode.create(url)
print(qr.terminal())
if have_tunnel and args.tunnel:
# pylint: disable=C0415
from . import tunnel
# Even if others said we could close, stay open if the user asked for a tunnel
closeNow = False
tunnel.Tunnel(interface, subnet=args.tunnel_net)
# if the user didn't ask for serial debugging output, we might want to exit after we've done our operation
if (not args.seriallog) and closeNow:
interface.close() # after running command then exit
except Exception as ex:
print(f"Aborting due to: {ex}")
interface.close() # close the connection now, so that our app exits
def onNode(node):
"""Callback invoked when the node DB changes"""
print(f"Node changed: {node}")
def subscribe():
"""Subscribe to the topics the user probably wants to see, prints output to stdout"""
pub.subscribe(onReceive, "meshtastic.receive")
# pub.subscribe(onConnection, "meshtastic.connection")
# We now call onConnected from main
# pub.subscribe(onConnected, "meshtastic.connection.established")
# pub.subscribe(onNode, "meshtastic.node")
def export_config(interface):
"""used in--export-config"""
owner = interface.getLongName()
channel_url = interface.localNode.getURL()
myinfo = interface.getMyNodeInfo()
pos = myinfo.get('position')
lat = None
lon = None
alt = None
if pos:
lat = pos.get('latitude')
lon = pos.get('longitude')
alt = pos.get('altitude')
config = "# start of Meshtastic configure yaml\n"
if owner:
config += f"owner: {owner}\n\n"
if channel_url:
config += f"channel_url: {channel_url}\n\n"
if lat or lon or alt:
config += "location:\n"
if lat:
config += f" lat: {lat}\n"
if lon:
config += f" lon: {lon}\n"
if alt:
config += f" alt: {alt}\n"
config += "\n"
preferences = f'{interface.localNode.radioConfig.preferences}'
prefs = preferences.splitlines()
if prefs:
config += "user_prefs:\n"
for pref in prefs:
config += f" {meshtastic.util.quoteBooleans(pref)}\n"
print(config)
return config
def common():
"""Shared code for all of our command line wrappers"""
our_globals = Globals.getInstance()
args = our_globals.get_args()
parser = our_globals.get_parser()
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO,
format='%(levelname)s file:%(filename)s %(funcName)s line:%(lineno)s %(message)s')
if len(sys.argv) == 1:
parser.print_help(sys.stderr)
meshtastic.util.our_exit("", 1)
else:
if args.support:
meshtastic.util.support_info()
meshtastic.util.our_exit("", 0)
if args.ch_index is not None:
channelIndex = int(args.ch_index)
our_globals.set_channel_index(channelIndex)
# Some commands require dest to be set, so we now use destOrAll/destOrLocal for more lenient commands
if not args.dest:
args.destOrAll = "^all"
args.destOrLocal = "^local"
else:
args.destOrAll = args.dest
args.destOrLocal = args.dest # FIXME, temp hack for debugging remove
if not args.seriallog:
if args.noproto:
args.seriallog = "stdout"
else:
args.seriallog = "none" # assume no debug output in this case
if args.deprecated is not None:
logging.error(
'This option has been deprecated, see help below for the correct replacement...')
parser.print_help(sys.stderr)
meshtastic.util.our_exit('', 1)
elif args.test:
result = meshtastic.test.testAll()
if not result:
meshtastic.util.our_exit("Warning: Test was not successful.")
else:
meshtastic.util.our_exit("Test was a success.", 0)
else:
if args.seriallog == "stdout":
logfile = sys.stdout
elif args.seriallog == "none":
args.seriallog = None
logging.debug("Not logging serial output")
logfile = None
else:
logging.info(f"Logging serial output to {args.seriallog}")
# Note: using "line buffering"
# pylint: disable=R1732
logfile = open(args.seriallog, 'w+',
buffering=1, encoding='utf8')
subscribe()
if args.ble:
client = meshtastic.ble_interface.BLEInterface(args.ble, debugOut=logfile, noProto=args.noproto)
elif args.host:
client = meshtastic.tcp_interface.TCPInterface(
args.host, debugOut=logfile, noProto=args.noproto)
else:
client = meshtastic.serial_interface.SerialInterface(
args.port, debugOut=logfile, noProto=args.noproto)
# We assume client is fully connected now
onConnected(client)
if args.noproto or (have_tunnel and args.tunnel): # loop until someone presses ctrlc
while True:
time.sleep(1000)
# don't call exit, background threads might be running still
# sys.exit(0)
def initParser():
"""Initialize the command line argument parsing."""
our_globals = Globals.getInstance()
parser = our_globals.get_parser()
args = our_globals.get_args()
parser.add_argument(
"--configure",
help="Specify a path to a yaml(.yml) file containing the desired settings for the connected device.",
action='append')
parser.add_argument(
"--export-config",
help="Export the configuration in yaml(.yml) format.",
action='store_true')
parser.add_argument(
"--port",
help="The port the Meshtastic device is connected to, i.e. /dev/ttyUSB0. If unspecified, we'll try to find it.",
default=None)
parser.add_argument(
"--host",
help="The hostname/ipaddr of the device to connect to (over TCP)",
default=None)
parser.add_argument(
"--seriallog",
help="Log device serial output to either 'stdout', 'none' or a filename to append to.")
parser.add_argument("--info", help="Read and display the radio config information",
action="store_true")
parser.add_argument("--nodes", help="Print Node List in a pretty formatted table",
action="store_true")
parser.add_argument("--qr", help="Display the QR code that corresponds to the current channel",
action="store_true")
parser.add_argument(
"--get", help="Get a preferences field. Use an invalid field such as '0' to get a list of all fields.", nargs=1, action='append')
parser.add_argument(
"--set", help="Set a preferences field", nargs=2, action='append')
parser.add_argument(
"--seturl", help="Set a channel URL", action="store")
parser.add_argument(
"--ch-index", help="Set the specified channel index. Channels start at 0 (0 is the PRIMARY channel).", action="store")
parser.add_argument(
"--ch-add", help="Add a secondary channel, you must specify a channel name", default=None)
parser.add_argument(
"--ch-del", help="Delete the ch-index channel", action='store_true')
parser.add_argument(
"--ch-enable", help="Enable the specified channel", action="store_true", dest="ch_enable", default=False)
# Note: We are doing a double negative here (Do we want to disable? If ch_disable==True, then disable.)
parser.add_argument(
"--ch-disable", help="Disable the specified channel", action="store_true", dest="ch_disable", default=False)
parser.add_argument(
"--ch-set", help="Set a channel parameter", nargs=2, action='append')
parser.add_argument(
"--ch-longslow", help="Change to the long-range and slow channel", action='store_true')
parser.add_argument(
"--ch-longfast", help="Change to the long-range and fast channel", action='store_true')
parser.add_argument(
"--ch-mediumslow", help="Change to the medium-range and slow channel", action='store_true')
parser.add_argument(
"--ch-mediumfast", help="Change to the medium-range and fast channel", action='store_true')
parser.add_argument(
"--ch-shortslow", help="Change to the short-range and slow channel", action='store_true')
parser.add_argument(
"--ch-shortfast", help="Change to the short-range and fast channel", action='store_true')
parser.add_argument(
"--set-owner", help="Set device owner name", action="store")
parser.add_argument(
"--set-team", help="Set team affiliation (an invalid team will list valid values)", action="store")
parser.add_argument(
"--set-ham", help="Set licensed Ham ID and turn off encryption", action="store")
parser.add_argument(
"--dest", help="The destination node id for any sent commands, if not set '^all' or '^local' is assumed as appropriate", default=None)
parser.add_argument(
"--sendtext", help="Send a text message. Can specify a destination '--dest' and/or channel index '--ch-index'.")
parser.add_argument(
"--sendping", help="Send a ping message (which requests a reply)", action="store_true")
parser.add_argument(
"--reboot", help="Tell the destination node to reboot", action="store_true")
parser.add_argument(
"--reply", help="Reply to received messages",
action="store_true")
parser.add_argument(
"--gpio-wrb", nargs=2, help="Set a particular GPIO # to 1 or 0", action='append')
parser.add_argument(
"--gpio-rd", help="Read from a GPIO mask (ex: '0x10')")
parser.add_argument(
"--gpio-watch", help="Start watching a GPIO mask for changes (ex: '0x10')")
parser.add_argument(
"--no-time", help="Suppress sending the current time to the mesh", action="store_true")
parser.add_argument(
"--setalt", help="Set device altitude (allows use without GPS)")
parser.add_argument(
"--setlat", help="Set device latitude (allows use without GPS)")
parser.add_argument(
"--setlon", help="Set device longitude (allows use without GPS)")
parser.add_argument(
"--pos-fields", help="Specify fields to send when sending a position. Use no argument for a list of valid values. "\
"Can pass multiple values as a space separated list like "\
"this: '--pos-fields POS_ALTITUDE POS_ALT_MSL'",
nargs="*", action="store")
parser.add_argument("--debug", help="Show API library debug log messages",
action="store_true")
parser.add_argument("--test", help="Run stress test against all connected Meshtastic devices",
action="store_true")
parser.add_argument("--ble", help="BLE mac address to connect to (BLE is not yet supported for this tool)",
default=None)
parser.add_argument("--noproto", help="Don't start the API, just function as a dumb serial terminal.",
action="store_true")
parser.add_argument('--setchan', dest='deprecated', nargs=2, action='append',
help='Deprecated, use "--ch-set param value" instead')
parser.add_argument('--set-router', dest='deprecated',
action='store_true', help='Deprecated, use "--set is_router true" instead')
parser.add_argument('--unset-router', dest='deprecated',
action='store_false', help='Deprecated, use "--set is_router false" instead')
if have_tunnel:
parser.add_argument('--tunnel',
action='store_true', help="Create a TUN tunnel device for forwarding IP packets over the mesh")
parser.add_argument(
"--subnet", dest='tunnel_net', help="Sets the local-end subnet address for the TUN IP bridge", default=None)
parser.set_defaults(deprecated=None)
parser.add_argument('--version', action='version',
version=f"{pkg_resources.require('meshtastic')[0].version}")
parser.add_argument(
"--support", action='store_true', help="Show support info (useful when troubleshooting an issue)")
args = parser.parse_args()
our_globals.set_args(args)
our_globals.set_parser(parser)
def main():
"""Perform command line meshtastic operations"""
our_globals = Globals.getInstance()
parser = argparse.ArgumentParser()
our_globals.set_parser(parser)
initParser()
common()
def tunnelMain():
"""Run a meshtastic IP tunnel"""
our_globals = Globals.getInstance()
parser = argparse.ArgumentParser()
our_globals.set_parser(parser)
initParser()
args = our_globals.get_args()
args.tunnel = True
our_globals.set_args(args)
common()
if __name__ == "__main__":
main()Global variables
var have_tunnel-
We only import the tunnel code if we are on a platform that can run it.
Functions
def common()-
Shared code for all of our command line wrappers
Expand source code
def common(): """Shared code for all of our command line wrappers""" our_globals = Globals.getInstance() args = our_globals.get_args() parser = our_globals.get_parser() logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO, format='%(levelname)s file:%(filename)s %(funcName)s line:%(lineno)s %(message)s') if len(sys.argv) == 1: parser.print_help(sys.stderr) meshtastic.util.our_exit("", 1) else: if args.support: meshtastic.util.support_info() meshtastic.util.our_exit("", 0) if args.ch_index is not None: channelIndex = int(args.ch_index) our_globals.set_channel_index(channelIndex) # Some commands require dest to be set, so we now use destOrAll/destOrLocal for more lenient commands if not args.dest: args.destOrAll = "^all" args.destOrLocal = "^local" else: args.destOrAll = args.dest args.destOrLocal = args.dest # FIXME, temp hack for debugging remove if not args.seriallog: if args.noproto: args.seriallog = "stdout" else: args.seriallog = "none" # assume no debug output in this case if args.deprecated is not None: logging.error( 'This option has been deprecated, see help below for the correct replacement...') parser.print_help(sys.stderr) meshtastic.util.our_exit('', 1) elif args.test: result = meshtastic.test.testAll() if not result: meshtastic.util.our_exit("Warning: Test was not successful.") else: meshtastic.util.our_exit("Test was a success.", 0) else: if args.seriallog == "stdout": logfile = sys.stdout elif args.seriallog == "none": args.seriallog = None logging.debug("Not logging serial output") logfile = None else: logging.info(f"Logging serial output to {args.seriallog}") # Note: using "line buffering" # pylint: disable=R1732 logfile = open(args.seriallog, 'w+', buffering=1, encoding='utf8') subscribe() if args.ble: client = meshtastic.ble_interface.BLEInterface(args.ble, debugOut=logfile, noProto=args.noproto) elif args.host: client = meshtastic.tcp_interface.TCPInterface( args.host, debugOut=logfile, noProto=args.noproto) else: client = meshtastic.serial_interface.SerialInterface( args.port, debugOut=logfile, noProto=args.noproto) # We assume client is fully connected now onConnected(client) if args.noproto or (have_tunnel and args.tunnel): # loop until someone presses ctrlc while True: time.sleep(1000) # don't call exit, background threads might be running still # sys.exit(0) def export_config(interface)-
used in–export-config
Expand source code
def export_config(interface): """used in--export-config""" owner = interface.getLongName() channel_url = interface.localNode.getURL() myinfo = interface.getMyNodeInfo() pos = myinfo.get('position') lat = None lon = None alt = None if pos: lat = pos.get('latitude') lon = pos.get('longitude') alt = pos.get('altitude') config = "# start of Meshtastic configure yaml\n" if owner: config += f"owner: {owner}\n\n" if channel_url: config += f"channel_url: {channel_url}\n\n" if lat or lon or alt: config += "location:\n" if lat: config += f" lat: {lat}\n" if lon: config += f" lon: {lon}\n" if alt: config += f" alt: {alt}\n" config += "\n" preferences = f'{interface.localNode.radioConfig.preferences}' prefs = preferences.splitlines() if prefs: config += "user_prefs:\n" for pref in prefs: config += f" {meshtastic.util.quoteBooleans(pref)}\n" print(config) return config def getPref(attributes, name)-
Get a channel or preferences value
Expand source code
def getPref(attributes, name): """Get a channel or preferences value""" objDesc = attributes.DESCRIPTOR field = objDesc.fields_by_name.get(name) if not field: print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not get it.") print(f"Choices in sorted order are:") names = [] for f in objDesc.fields: names.append(f'{f.name}') for temp_name in sorted(names): print(f" {temp_name}") return # okay - try to read the value try: try: val = getattr(attributes, name) except TypeError: # The getter didn't like our arg type guess try again as a string val = getattr(attributes, name) # succeeded! print(f"{name}: {str(val)}") except Exception as ex: print(f"Can't get {name} due to {ex}") def initParser()-
Initialize the command line argument parsing.
Expand source code
def initParser(): """Initialize the command line argument parsing.""" our_globals = Globals.getInstance() parser = our_globals.get_parser() args = our_globals.get_args() parser.add_argument( "--configure", help="Specify a path to a yaml(.yml) file containing the desired settings for the connected device.", action='append') parser.add_argument( "--export-config", help="Export the configuration in yaml(.yml) format.", action='store_true') parser.add_argument( "--port", help="The port the Meshtastic device is connected to, i.e. /dev/ttyUSB0. If unspecified, we'll try to find it.", default=None) parser.add_argument( "--host", help="The hostname/ipaddr of the device to connect to (over TCP)", default=None) parser.add_argument( "--seriallog", help="Log device serial output to either 'stdout', 'none' or a filename to append to.") parser.add_argument("--info", help="Read and display the radio config information", action="store_true") parser.add_argument("--nodes", help="Print Node List in a pretty formatted table", action="store_true") parser.add_argument("--qr", help="Display the QR code that corresponds to the current channel", action="store_true") parser.add_argument( "--get", help="Get a preferences field. Use an invalid field such as '0' to get a list of all fields.", nargs=1, action='append') parser.add_argument( "--set", help="Set a preferences field", nargs=2, action='append') parser.add_argument( "--seturl", help="Set a channel URL", action="store") parser.add_argument( "--ch-index", help="Set the specified channel index. Channels start at 0 (0 is the PRIMARY channel).", action="store") parser.add_argument( "--ch-add", help="Add a secondary channel, you must specify a channel name", default=None) parser.add_argument( "--ch-del", help="Delete the ch-index channel", action='store_true') parser.add_argument( "--ch-enable", help="Enable the specified channel", action="store_true", dest="ch_enable", default=False) # Note: We are doing a double negative here (Do we want to disable? If ch_disable==True, then disable.) parser.add_argument( "--ch-disable", help="Disable the specified channel", action="store_true", dest="ch_disable", default=False) parser.add_argument( "--ch-set", help="Set a channel parameter", nargs=2, action='append') parser.add_argument( "--ch-longslow", help="Change to the long-range and slow channel", action='store_true') parser.add_argument( "--ch-longfast", help="Change to the long-range and fast channel", action='store_true') parser.add_argument( "--ch-mediumslow", help="Change to the medium-range and slow channel", action='store_true') parser.add_argument( "--ch-mediumfast", help="Change to the medium-range and fast channel", action='store_true') parser.add_argument( "--ch-shortslow", help="Change to the short-range and slow channel", action='store_true') parser.add_argument( "--ch-shortfast", help="Change to the short-range and fast channel", action='store_true') parser.add_argument( "--set-owner", help="Set device owner name", action="store") parser.add_argument( "--set-team", help="Set team affiliation (an invalid team will list valid values)", action="store") parser.add_argument( "--set-ham", help="Set licensed Ham ID and turn off encryption", action="store") parser.add_argument( "--dest", help="The destination node id for any sent commands, if not set '^all' or '^local' is assumed as appropriate", default=None) parser.add_argument( "--sendtext", help="Send a text message. Can specify a destination '--dest' and/or channel index '--ch-index'.") parser.add_argument( "--sendping", help="Send a ping message (which requests a reply)", action="store_true") parser.add_argument( "--reboot", help="Tell the destination node to reboot", action="store_true") parser.add_argument( "--reply", help="Reply to received messages", action="store_true") parser.add_argument( "--gpio-wrb", nargs=2, help="Set a particular GPIO # to 1 or 0", action='append') parser.add_argument( "--gpio-rd", help="Read from a GPIO mask (ex: '0x10')") parser.add_argument( "--gpio-watch", help="Start watching a GPIO mask for changes (ex: '0x10')") parser.add_argument( "--no-time", help="Suppress sending the current time to the mesh", action="store_true") parser.add_argument( "--setalt", help="Set device altitude (allows use without GPS)") parser.add_argument( "--setlat", help="Set device latitude (allows use without GPS)") parser.add_argument( "--setlon", help="Set device longitude (allows use without GPS)") parser.add_argument( "--pos-fields", help="Specify fields to send when sending a position. Use no argument for a list of valid values. "\ "Can pass multiple values as a space separated list like "\ "this: '--pos-fields POS_ALTITUDE POS_ALT_MSL'", nargs="*", action="store") parser.add_argument("--debug", help="Show API library debug log messages", action="store_true") parser.add_argument("--test", help="Run stress test against all connected Meshtastic devices", action="store_true") parser.add_argument("--ble", help="BLE mac address to connect to (BLE is not yet supported for this tool)", default=None) parser.add_argument("--noproto", help="Don't start the API, just function as a dumb serial terminal.", action="store_true") parser.add_argument('--setchan', dest='deprecated', nargs=2, action='append', help='Deprecated, use "--ch-set param value" instead') parser.add_argument('--set-router', dest='deprecated', action='store_true', help='Deprecated, use "--set is_router true" instead') parser.add_argument('--unset-router', dest='deprecated', action='store_false', help='Deprecated, use "--set is_router false" instead') if have_tunnel: parser.add_argument('--tunnel', action='store_true', help="Create a TUN tunnel device for forwarding IP packets over the mesh") parser.add_argument( "--subnet", dest='tunnel_net', help="Sets the local-end subnet address for the TUN IP bridge", default=None) parser.set_defaults(deprecated=None) parser.add_argument('--version', action='version', version=f"{pkg_resources.require('meshtastic')[0].version}") parser.add_argument( "--support", action='store_true', help="Show support info (useful when troubleshooting an issue)") args = parser.parse_args() our_globals.set_args(args) our_globals.set_parser(parser) def main()-
Perform command line meshtastic operations
Expand source code
def main(): """Perform command line meshtastic operations""" our_globals = Globals.getInstance() parser = argparse.ArgumentParser() our_globals.set_parser(parser) initParser() common() def onConnected(interface)-
Callback invoked when we connect to a radio
Expand source code
def onConnected(interface): """Callback invoked when we connect to a radio""" closeNow = False # Should we drop the connection after we finish? try: our_globals = Globals.getInstance() args = our_globals.get_args() print("Connected to radio") def getNode(): """This operation could be expensive, so we try to cache the results""" targetNode = our_globals.get_target_node() if not targetNode: targetNode = interface.getNode(args.destOrLocal) our_globals.set_target_node(targetNode) return targetNode if args.setlat or args.setlon or args.setalt: closeNow = True alt = 0 lat = 0.0 lon = 0.0 prefs = interface.localNode.radioConfig.preferences if args.setalt: alt = int(args.setalt) prefs.fixed_position = True print(f"Fixing altitude at {alt} meters") if args.setlat: lat = float(args.setlat) prefs.fixed_position = True print(f"Fixing latitude at {lat} degrees") if args.setlon: lon = float(args.setlon) prefs.fixed_position = True print(f"Fixing longitude at {lon} degrees") print("Setting device position") # can include lat/long/alt etc: latitude = 37.5, longitude = -122.1 interface.sendPosition(lat, lon, alt) interface.localNode.writeConfig() elif not args.no_time: # We normally provide a current time to the mesh when we connect interface.sendPosition() if args.set_owner: closeNow = True print(f"Setting device owner to {args.set_owner}") getNode().setOwner(args.set_owner) if args.pos_fields: # If --pos-fields invoked with args, set position fields closeNow = True prefs = getNode().radioConfig.preferences allFields = 0 try: for field in args.pos_fields: v_field = radioconfig_pb2.PositionFlags.Value(field) allFields |= v_field except ValueError: print("ERROR: supported position fields are:") print(radioconfig_pb2.PositionFlags.keys()) print("If no fields are specified, will read and display current value.") else: print(f"Setting position fields to {allFields}") setPref(prefs, 'position_flags', ('%d' % allFields)) print("Writing modified preferences to device") getNode().writeConfig() elif args.pos_fields is not None: # If --pos-fields invoked without args, read and display current value closeNow = True prefs = getNode().radioConfig.preferences fieldNames = [] for bit in radioconfig_pb2.PositionFlags.values(): if prefs.position_flags & bit: fieldNames.append(radioconfig_pb2.PositionFlags.Name(bit)) print(' '.join(fieldNames)) if args.set_team: closeNow = True try: v_team = meshtastic.mesh_pb2.Team.Value(args.set_team.upper()) except ValueError: v_team = 0 print(f"ERROR: Team \'{args.set_team}\' not found.") print("Try a team name from the sorted list below, or use 'CLEAR' for unaffiliated:") print(sorted(meshtastic.mesh_pb2.Team.keys())) else: print(f"Setting team to {meshtastic.mesh_pb2.Team.Name(v_team)}") getNode().setOwner(team=v_team) if args.set_ham: closeNow = True print(f"Setting Ham ID to {args.set_ham} and turning off encryption") getNode().setOwner(args.set_ham, is_licensed=True) # Must turn off encryption on primary channel getNode().turnOffEncryptionOnPrimaryChannel() if args.reboot: closeNow = True getNode().reboot() if args.sendtext: closeNow = True channelIndex = 0 if args.ch_index is not None: channelIndex = int(args.ch_index) ch = getNode().getChannelByChannelIndex(channelIndex) if ch and ch.role != channel_pb2.Channel.Role.DISABLED: print(f"Sending text message {args.sendtext} to {args.destOrAll} on channelIndex:{channelIndex}") interface.sendText(args.sendtext, args.destOrAll, wantAck=True, channelIndex=channelIndex) else: meshtastic.util.our_exit(f"Warning: {channelIndex} is not a valid channel. Channel must not be DISABLED.") if args.sendping: payload = str.encode("test string") print(f"Sending ping message to {args.destOrAll}") interface.sendData(payload, args.destOrAll, portNum=portnums_pb2.PortNum.REPLY_APP, wantAck=True, wantResponse=True) if args.gpio_wrb or args.gpio_rd or args.gpio_watch: rhc = remote_hardware.RemoteHardwareClient(interface) if args.gpio_wrb: bitmask = 0 bitval = 0 for wrpair in (args.gpio_wrb or []): bitmask |= 1 << int(wrpair[0]) bitval |= int(wrpair[1]) << int(wrpair[0]) print(f"Writing GPIO mask 0x{bitmask:x} with value 0x{bitval:x} to {args.dest}") rhc.writeGPIOs(args.dest, bitmask, bitval) closeNow = True if args.gpio_rd: bitmask = int(args.gpio_rd, 16) print(f"Reading GPIO mask 0x{bitmask:x} from {args.dest}") interface.mask = bitmask rhc.readGPIOs(args.dest, bitmask, None) if not interface.noProto: # wait up to X seconds for a response for _ in range(10): time.sleep(1) if interface.gotResponse: break logging.debug(f'end of gpio_rd') if args.gpio_watch: bitmask = int(args.gpio_watch, 16) print(f"Watching GPIO mask 0x{bitmask:x} from {args.dest}. Press ctrl-c to exit") while True: rhc.watchGPIOs(args.dest, bitmask) time.sleep(1) # handle settings if args.set: closeNow = True prefs = getNode().radioConfig.preferences # Handle the int/float/bool arguments for pref in args.set: setPref(prefs, pref[0], pref[1]) print("Writing modified preferences to device") getNode().writeConfig() if args.configure: with open(args.configure[0], encoding='utf8') as file: configuration = yaml.safe_load(file) closeNow = True if 'owner' in configuration: print(f"Setting device owner to {configuration['owner']}") getNode().setOwner(configuration['owner']) if 'channel_url' in configuration: print("Setting channel url to", configuration['channel_url']) getNode().setURL(configuration['channel_url']) if 'location' in configuration: alt = 0 lat = 0.0 lon = 0.0 prefs = interface.localNode.radioConfig.preferences if 'alt' in configuration['location']: alt = int(configuration['location']['alt']) prefs.fixed_position = True print(f"Fixing altitude at {alt} meters") if 'lat' in configuration['location']: lat = float(configuration['location']['lat']) prefs.fixed_position = True print(f"Fixing latitude at {lat} degrees") if 'lon' in configuration['location']: lon = float(configuration['location']['lon']) prefs.fixed_position = True print(f"Fixing longitude at {lon} degrees") print("Setting device position") interface.sendPosition(lat, lon, alt) interface.localNode.writeConfig() if 'user_prefs' in configuration: prefs = getNode().radioConfig.preferences for pref in configuration['user_prefs']: setPref(prefs, pref, str(configuration['user_prefs'][pref])) print("Writing modified preferences to device") getNode().writeConfig() if args.export_config: # export the configuration (the opposite of '--configure') closeNow = True export_config(interface) if args.seturl: closeNow = True getNode().setURL(args.seturl) # handle changing channels if args.ch_add: closeNow = True if len(args.ch_add) > 10: meshtastic.util.our_exit("Warning: Channel name must be shorter. Channel not added.") n = getNode() ch = n.getChannelByName(args.ch_add) if ch: meshtastic.util.our_exit(f"Warning: This node already has a '{args.ch_add}' channel. No changes were made.") else: # get the first channel that is disabled (i.e., available) ch = n.getDisabledChannel() if not ch: meshtastic.util.our_exit("Warning: No free channels were found") chs = channel_pb2.ChannelSettings() chs.psk = meshtastic.util.genPSK256() chs.name = args.ch_add ch.settings.CopyFrom(chs) ch.role = channel_pb2.Channel.Role.SECONDARY print(f"Writing modified channels to device") n.writeChannel(ch.index) if args.ch_del: closeNow = True channelIndex = our_globals.get_channel_index() if channelIndex is None: meshtastic.util.our_exit("Warning: Need to specify '--ch-index' for '--ch-del'.", 1) else: if channelIndex == 0: meshtastic.util.our_exit("Warning: Cannot delete primary channel.", 1) else: print(f"Deleting channel {channelIndex}") ch = getNode().deleteChannel(channelIndex) ch_changes = [args.ch_longslow, args.ch_longfast, args.ch_mediumslow, args.ch_mediumfast, args.ch_shortslow, args.ch_shortfast] any_primary_channel_changes = any(x for x in ch_changes) if args.ch_set or any_primary_channel_changes or args.ch_enable or args.ch_disable: closeNow = True channelIndex = our_globals.get_channel_index() if channelIndex is None: if any_primary_channel_changes: # we assume that they want the primary channel if they're setting range values channelIndex = 0 else: meshtastic.util.our_exit("Warning: Need to specify '--ch-index'.", 1) ch = getNode().channels[channelIndex] if any_primary_channel_changes or args.ch_enable or args.ch_disable: if channelIndex == 0 and not any_primary_channel_changes: meshtastic.util.our_exit("Warning: Cannot enable/disable PRIMARY channel.") if channelIndex != 0: if any_primary_channel_changes: meshtastic.util.our_exit("Warning: Standard channel settings can only be applied to the PRIMARY channel") enable = True # default to enable if args.ch_enable: enable = True if args.ch_disable: enable = False def setSimpleChannel(modem_config): """Set one of the simple modem_config only based channels""" # Completely new channel settings chs = channel_pb2.ChannelSettings() chs.modem_config = modem_config chs.psk = bytes([1]) # Use default channel psk 1 ch.settings.CopyFrom(chs) # handle the simple channel set commands if args.ch_longslow: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr48Sf4096) if args.ch_longfast: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw31_25Cr48Sf512) if args.ch_mediumslow: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr46Sf2048) if args.ch_mediumfast: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw250Cr47Sf1024) if args.ch_shortslow: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw125Cr45Sf128) if args.ch_shortfast: setSimpleChannel(channel_pb2.ChannelSettings.ModemConfig.Bw500Cr45Sf128) # Handle the channel settings for pref in (args.ch_set or []): if pref[0] == "psk": ch.settings.psk = meshtastic.util.fromPSK(pref[1]) else: setPref(ch.settings, pref[0], pref[1]) enable = True # If we set any pref, assume the user wants to enable the channel if enable: ch.role = channel_pb2.Channel.Role.PRIMARY if ( channelIndex == 0) else channel_pb2.Channel.Role.SECONDARY else: ch.role = channel_pb2.Channel.Role.DISABLED print(f"Writing modified channels to device") getNode().writeChannel(channelIndex) if args.info: print("") if not args.dest: # If we aren't trying to talk to our local node, don't show it interface.showInfo() print("") getNode().showInfo() closeNow = True # FIXME, for now we leave the link up while talking to remote nodes print("") if args.get: closeNow = True prefs = getNode().radioConfig.preferences # Handle the int/float/bool arguments for pref in args.get: getPref(prefs, pref[0]) print("Completed getting preferences") if args.nodes: closeNow = True interface.showNodes() if args.qr: closeNow = True url = interface.localNode.getURL(includeAll=False) print(f"Primary channel URL {url}") qr = pyqrcode.create(url) print(qr.terminal()) if have_tunnel and args.tunnel: # pylint: disable=C0415 from . import tunnel # Even if others said we could close, stay open if the user asked for a tunnel closeNow = False tunnel.Tunnel(interface, subnet=args.tunnel_net) # if the user didn't ask for serial debugging output, we might want to exit after we've done our operation if (not args.seriallog) and closeNow: interface.close() # after running command then exit except Exception as ex: print(f"Aborting due to: {ex}") interface.close() # close the connection now, so that our app exits def onConnection(interface, topic=pubsub.core.callables.AUTO_TOPIC)-
Callback invoked when we connect/disconnect from a radio
Expand source code
def onConnection(interface, topic=pub.AUTO_TOPIC): """Callback invoked when we connect/disconnect from a radio""" print(f"Connection changed: {topic.getName()}") def onNode(node)-
Callback invoked when the node DB changes
Expand source code
def onNode(node): """Callback invoked when the node DB changes""" print(f"Node changed: {node}") def onReceive(packet, interface)-
Callback invoked when a packet arrives
Expand source code
def onReceive(packet, interface): """Callback invoked when a packet arrives""" our_globals = Globals.getInstance() args = our_globals.get_args() try: d = packet.get('decoded') logging.debug(f'in onReceive() d:{d}') # Exit once we receive a reply if args and args.sendtext and packet["to"] == interface.myInfo.my_node_num and d["portnum"] == portnums_pb2.PortNum.TEXT_MESSAGE_APP: interface.close() # after running command then exit # Reply to every received message with some stats if args and args.reply: msg = d.get('text') if msg: rxSnr = packet['rxSnr'] hopLimit = packet['hopLimit'] print(f"message: {msg}") reply = "got msg \'{}\' with rxSnr: {} and hopLimit: {}".format(msg, rxSnr, hopLimit) print("Sending reply: ", reply) interface.sendText(reply) except Exception as ex: print(ex) def setPref(attributes, name, valStr)-
Set a channel or preferences value
Expand source code
def setPref(attributes, name, valStr): """Set a channel or preferences value""" objDesc = attributes.DESCRIPTOR field = objDesc.fields_by_name.get(name) if not field: print(f"{attributes.__class__.__name__} does not have an attribute called {name}, so you can not set it.") print(f"Choices in sorted order are:") names = [] for f in objDesc.fields: names.append(f'{f.name}') for temp_name in sorted(names): print(f" {temp_name}") return val = meshtastic.util.fromStr(valStr) enumType = field.enum_type # pylint: disable=C0123 if enumType and type(val) == str: # We've failed so far to convert this string into an enum, try to find it by reflection e = enumType.values_by_name.get(val) if e: val = e.number else: print(f"{name} does not have an enum called {val}, so you can not set it.") print(f"Choices in sorted order are:") names = [] for f in enumType.values: names.append(f'{f.name}') for temp_name in sorted(names): print(f" {temp_name}") return # okay - try to read the value try: try: setattr(attributes, name, val) except TypeError: # The setter didn't like our arg type guess try again as a string setattr(attributes, name, valStr) # succeeded! print(f"Set {name} to {valStr}") except Exception as ex: print(f"Can't set {name} due to {ex}") def subscribe()-
Subscribe to the topics the user probably wants to see, prints output to stdout
Expand source code
def subscribe(): """Subscribe to the topics the user probably wants to see, prints output to stdout""" pub.subscribe(onReceive, "meshtastic.receive") # pub.subscribe(onConnection, "meshtastic.connection") # We now call onConnected from main # pub.subscribe(onConnected, "meshtastic.connection.established") # pub.subscribe(onNode, "meshtastic.node") def tunnelMain()-
Run a meshtastic IP tunnel
Expand source code
def tunnelMain(): """Run a meshtastic IP tunnel""" our_globals = Globals.getInstance() parser = argparse.ArgumentParser() our_globals.set_parser(parser) initParser() args = our_globals.get_args() args.tunnel = True our_globals.set_args(args) common()