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()
+