mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2025-12-31 03:47:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #126 from mkinney/add_pylint

Browse Source 
Add pylint
This commit is contained in:
Jm Casler
2021-11-30 14:33:57 -08:00
committed by GitHub
parent dfe3989cd8 035ec09b20
commit 04c5cb1f47
9 changed files with 185 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
.pylintrc Normal file
View File

@@ -0,0 +1,83 @@
# pylint configuration file
#
# Note: "pylint --generate-rcfile" is helpful to see what values to add to this file
[MASTER]
# Add files or directories matching the regex patterns to the blacklist. The
# regex matches against base names, not paths.
ignore-patterns=mqtt_pb2.py,channel_pb2.py,environmental_measurement_pb2.py,admin_pb2.py,radioconfig_pb2.py,deviceonly_pb2.py,apponly_pb2.py,remote_hardware_pb2.py,portnums_pb2.py,mesh_pb2.py
[MESSAGES CONTROL]
# Disable the message, report, category or checker with the given id(s). You
# can either give multiple identifiers separated by comma (,) or put this
# option multiple times (only on the command line, not in the configuration
# file where it should appear only once).You can also use "--disable=all" to
# disable everything first and then reenable specific checks. For example, if
# you want to run only the similarities checker, you can use "--disable=all
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W"
#
disable=invalid-name,fixme,logging-fstring-interpolation,too-many-statements,too-many-branches,too-many-locals,no-member,f-string-without-interpolation,protected-access,no-self-use,pointless-string-statement,too-few-public-methods,consider-using-f-string
[BASIC]
# Good variable names which should always be accepted, separated by a comma
good-names=i,j,k,ex,Run,_
# Bad variable names which should always be refused, separated by a comma
bad-names=foo,bar,baz,toto,tutu,tata
[FORMAT]
# Maximum number of characters on a single line.
max-line-length=150
# Maximum number of lines in a module
max-module-lines=1200
[REFACTORING]
# Maximum number of nested blocks for function / method body
max-nested-blocks=10
[MISCELLANEOUS]
# List of note tags to take in consideration, separated by a comma.
notes=FIXME,fixme,XXX,TODO
[SIMILARITIES]
# Minimum lines number of a similarity.
min-similarity-lines=30
# Ignore comments when computing similarities.
ignore-comments=yes
# Ignore docstrings when computing similarities.
ignore-docstrings=yes
# Ignore imports when computing similarities.
ignore-imports=yes
[DESIGN]
# Maximum number of arguments for function / method.
max-args=10
# Maximum number of attributes for a class (see R0902).
max-attributes=20

7
README.md
View File

@@ -156,5 +156,10 @@ If you need to build a new release you'll need:
```
apt install pandoc
sudo pip3 install markdown pdoc3 webencodings pyparsing twine autopep8
sudo pip3 install markdown pdoc3 webencodings pyparsing twine autopep8 pylint
```
To lint, run:
```
pylint meshtastic
```

79
meshtastic/__init__.py
View File

@@ -28,10 +28,10 @@ type of packet, you should subscribe to the full topic name. If you want to see
- meshtastic.receive.data.portnum(packet) (where portnum is an integer or well known PortNum enum)
- meshtastic.node.updated(node = NodeInfo) - published when a node in the DB changes (appears, location changed, username changed, etc...)
We receive position, user, or data packets from the mesh. You probably only care about meshtastic.receive.data. The first argument for
that publish will be the packet. Text or binary data packets (from sendData or sendText) will both arrive this way. If you print packet
you'll see the fields in the dictionary. decoded.data.payload will contain the raw bytes that were sent. If the packet was sent with
sendText, decoded.data.text will **also** be populated with the decoded string. For ASCII these two strings will be the same, but for
We receive position, user, or data packets from the mesh. You probably only care about meshtastic.receive.data. The first argument for
that publish will be the packet. Text or binary data packets (from sendData or sendText) will both arrive this way. If you print packet
you'll see the fields in the dictionary. decoded.data.payload will contain the raw bytes that were sent. If the packet was sent with
sendText, decoded.data.text will **also** be populated with the decoded string. For ASCII these two strings will be the same, but for
unicode scripts they can be different.
# Example Usage
@@ -55,30 +55,30 @@ interface = meshtastic.SerialInterface()
"""
import pygatt
import google.protobuf.json_format
import serial
import threading
import base64
import logging
import sys
import os
import platform
import random
import socket
import sys
import stat
import threading
import traceback
import time
import base64
import platform
import socket
from datetime import datetime
from typing import *
import serial
import timeago
import os
import stat
from . import mesh_pb2, portnums_pb2, apponly_pb2, admin_pb2, environmental_measurement_pb2, remote_hardware_pb2, channel_pb2, radioconfig_pb2, util
from .util import fixme, catchAndIgnore, stripnl, DeferredExecution, Timeout
from .node import Node
import google.protobuf.json_format
import pygatt
from pubsub import pub
from dotmap import DotMap
from datetime import datetime
from tabulate import tabulate
from typing import *
from google.protobuf.json_format import MessageToJson
from .util import fixme, catchAndIgnore, stripnl, DeferredExecution, Timeout
from .node import Node
from . import mesh_pb2, portnums_pb2, apponly_pb2, admin_pb2, environmental_measurement_pb2, remote_hardware_pb2, channel_pb2, radioconfig_pb2, util
START1 = 0x94
START2 = 0xc3
@@ -272,7 +272,7 @@ class MeshInterface:
wantResponse=wantResponse,
hopLimit=hopLimit,
onResponse=onResponse,
channelIndex=channelIndex);
channelIndex=channelIndex)
def sendData(self, data, destinationId=BROADCAST_ADDR,
portNum=portnums_pb2.PortNum.PRIVATE_APP, wantAck=False,
@@ -288,7 +288,8 @@ class MeshInterface:
portNum -- the application portnum (similar to IP port numbers) of the destination, see portnums.proto for a list
wantAck -- True if you want the message sent in a reliable manner (with retries and ack/nak provided for delivery)
wantResponse -- True if you want the service on the other side to send an application layer response
onResponse -- A closure of the form funct(packet), that will be called when a response packet arrives (or the transaction is NAKed due to non receipt)
onResponse -- A closure of the form funct(packet), that will be called when a response packet arrives
(or the transaction is NAKed due to non receipt)
Returns the sent packet. The id field will be populated in this packet and can be used to track future message acks/naks.
"""
@@ -326,13 +327,13 @@ class MeshInterface:
Returns the sent packet. The id field will be populated in this packet and can be used to track future message acks/naks.
"""
p = mesh_pb2.Position()
if(latitude != 0.0):
if latitude != 0.0:
p.latitude_i = int(latitude / 1e-7)
if(longitude != 0.0):
if longitude != 0.0:
p.longitude_i = int(longitude / 1e-7)
if(altitude != 0):
if altitude != 0:
p.altitude = int(altitude)
if timeSec == 0:
@@ -353,7 +354,7 @@ class MeshInterface:
"""Send a MeshPacket to the specified node (or if unspecified, broadcast).
You probably don't want this - use sendData instead.
Returns the sent packet. The id field will be populated in this packet and
Returns the sent packet. The id field will be populated in this packet and
can be used to track future message acks/naks.
"""
@@ -402,23 +403,27 @@ class MeshInterface:
raise Exception("Timed out waiting for interface config")
def getMyNodeInfo(self):
"""Get info about my node."""
if self.myInfo is None:
return None
return self.nodesByNum.get(self.myInfo.my_node_num)
def getMyUser(self):
"""Get user"""
nodeInfo = self.getMyNodeInfo()
if nodeInfo is not None:
return nodeInfo.get('user')
return None
def getLongName(self):
"""Get long name"""
user = self.getMyUser()
if user is not None:
return user.get('longName', None)
return None
def getShortName(self):
"""Get short name"""
user = self.getMyUser()
if user is not None:
return user.get('shortName', None)
@@ -752,7 +757,7 @@ class StreamInterface(MeshInterface):
"""Interface class for meshtastic devices over a stream link (serial, TCP, etc)"""
def __init__(self, debugOut=None, noProto=False, connectNow=True):
"""Constructor, opens a connection to self.stream
"""Constructor, opens a connection to self.stream
Keyword Arguments:
devPath {string} -- A filepath to a device, i.e. /dev/ttyUSB0 (default: {None})
@@ -932,7 +937,7 @@ class SerialInterface(StreamInterface):
# rts=False Needed to prevent TBEAMs resetting on OSX, because rts is connected to reset
self.stream.port = devPath
# HACK: If the platform driving the serial port is unable to leave the RTS pin in high-impedance
# mode, set RTS to false so that the device platform won't be reset spuriously.
# Linux does this properly, so don't apply this hack on Linux (because it makes the reset button not work).
@@ -942,30 +947,30 @@ class SerialInterface(StreamInterface):
StreamInterface.__init__(
self, debugOut=debugOut, noProto=noProto, connectNow=connectNow)
"""true if platform driving the serial port is Windows Subsystem for Linux 1."""
def _isWsl1(self):
# WSL1 identifies itself as Linux, but has a special char device at /dev/lxss for use with session control,
# e.g. /init. We should treat WSL1 as Windows for the RTS-driving hack because the underlying platfrom
# WSL1 identifies itself as Linux, but has a special char device at /dev/lxss for use with session control,
# e.g. /init. We should treat WSL1 as Windows for the RTS-driving hack because the underlying platfrom
# serial driver for the CP21xx still exhibits the buggy behavior.
# WSL2 is not covered here, as it does not (as of 2021-May-25) support the appropriate functionality to
# WSL2 is not covered here, as it does not (as of 2021-May-25) support the appropriate functionality to
# share or pass-through serial ports.
try:
# Claims to be Linux, but has /dev/lxss; must be WSL 1
return platform.system() == 'Linux' and stat.S_ISCHR(os.stat('/dev/lxss').st_mode);
return platform.system() == 'Linux' and stat.S_ISCHR(os.stat('/dev/lxss').st_mode)
except:
# Couldn't stat /dev/lxss special device; not WSL1
return False;
return False
def _hostPlatformAlwaysDrivesUartRts(self):
# OS-X/Windows seems to have a bug in its CP21xx serial drivers. It ignores that we asked for no RTSCTS
# control and will always drive RTS either high or low (rather than letting the CP102 leave
# it as an open-collector floating pin).
# TODO: When WSL2 supports USB passthrough, this will get messier. If/when WSL2 gets virtual serial
# ports that "share" the Windows serial port (and thus the Windows drivers), this code will need to be
# TODO: When WSL2 supports USB passthrough, this will get messier. If/when WSL2 gets virtual serial
# ports that "share" the Windows serial port (and thus the Windows drivers), this code will need to be
# updated to reflect that as well -- or if T-Beams get made with an alternate USB to UART bridge that has
# a less buggy driver.
return platform.system() != 'Linux' or self._isWsl1();
return platform.system() != 'Linux' or self._isWsl1()
class TCPInterface(StreamInterface):
"""Interface class for meshtastic devices over a TCP link"""

25
meshtastic/__main__.py
View File

@@ -1,21 +1,18 @@
#!python3
""" Main Meshtastic
"""
import argparse
import platform
import logging
import sys
import codecs
import time
import base64
import os
from . import SerialInterface, TCPInterface, BLEInterface, test, remote_hardware
from pubsub import pub
from . import mesh_pb2, portnums_pb2, channel_pb2
from .util import stripnl
import google.protobuf.json_format
import pyqrcode
import traceback
import pkg_resources
from . import SerialInterface, TCPInterface, BLEInterface, test, remote_hardware
from . import portnums_pb2, channel_pb2
"""We only import the tunnel code if we are on a platform that can run it"""
have_tunnel = platform.system() == 'Linux'
@@ -65,11 +62,12 @@ falseTerms = {"f", "false", "no"}
def genPSK256():
"""Generate a random preshared key"""
return os.urandom(32)
def fromPSK(valstr):
"""A special version of fromStr that assumes the user is trying to set a PSK.
"""A special version of fromStr that assumes the user is trying to set a PSK.
In that case we also allow "none", "default" or "random" (to have python generate one), or simpleN
"""
if valstr == "random":
@@ -93,9 +91,9 @@ def fromStr(valstr):
Args:
valstr (string): A user provided string
"""
if(len(valstr) == 0): # Treat an emptystring as an empty bytes
if len(valstr) == 0: # Treat an emptystring as an empty bytes
val = bytes()
elif(valstr.startswith('0x')):
elif valstr.startswith('0x'):
# if needed convert to string with asBytes.decode('utf-8')
val = bytes.fromhex(valstr[2:])
elif valstr in trueTerms:
@@ -134,7 +132,7 @@ def getPref(attributes, name):
try:
try:
val = getattr(attributes, name)
except TypeError as ex:
except TypeError:
# The getter didn't like our arg type guess try again as a string
val = getattr(attributes, name)
@@ -175,7 +173,7 @@ def setPref(attributes, name, valStr):
try:
try:
setattr(attributes, name, val)
except TypeError as ex:
except TypeError:
# The setter didn't like our arg type guess try again as a string
setattr(attributes, name, valStr)
@@ -479,7 +477,7 @@ def common():
else:
args.seriallog = "none" # assume no debug output in this case
if args.deprecated != None:
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)
@@ -520,6 +518,7 @@ def common():
def initParser():
""" Initialize the command line argument parsing."""
global parser, args
parser.add_argument(

30
meshtastic/node.py
View File

@@ -28,10 +28,10 @@ type of packet, you should subscribe to the full topic name. If you want to see
- meshtastic.receive.data.portnum(packet) (where portnum is an integer or well known PortNum enum)
- meshtastic.node.updated(node = NodeInfo) - published when a node in the DB changes (appears, location changed, username changed, etc...)
We receive position, user, or data packets from the mesh. You probably only care about meshtastic.receive.data. The first argument for
that publish will be the packet. Text or binary data packets (from sendData or sendText) will both arrive this way. If you print packet
you'll see the fields in the dictionary. decoded.data.payload will contain the raw bytes that were sent. If the packet was sent with
sendText, decoded.data.text will **also** be populated with the decoded string. For ASCII these two strings will be the same, but for
We receive position, user, or data packets from the mesh. You probably only care about meshtastic.receive.data. The first argument for
that publish will be the packet. Text or binary data packets (from sendData or sendText) will both arrive this way. If you print packet
you'll see the fields in the dictionary. decoded.data.payload will contain the raw bytes that were sent. If the packet was sent with
sendText, decoded.data.text will **also** be populated with the decoded string. For ASCII these two strings will be the same, but for
unicode scripts they can be different.
# Example Usage
@@ -55,24 +55,12 @@ interface = meshtastic.SerialInterface()
"""
import pygatt
import google.protobuf.json_format
import serial
import threading
import logging
import sys
import random
import traceback
import time
import base64
import platform
import socket
from . import mesh_pb2, portnums_pb2, apponly_pb2, admin_pb2, environmental_measurement_pb2, remote_hardware_pb2, channel_pb2, radioconfig_pb2, util
from .util import fixme, catchAndIgnore, stripnl, DeferredExecution, Timeout
from pubsub import pub
from dotmap import DotMap
from typing import *
from google.protobuf.json_format import MessageToJson
from . import portnums_pb2, apponly_pb2, admin_pb2, channel_pb2
from .util import stripnl, Timeout
@@ -142,7 +130,7 @@ class Node:
def writeConfig(self):
"""Write the current (edited) radioConfig to the device"""
if self.radioConfig == None:
if self.radioConfig is None:
raise Exception("No RadioConfig has been read")
p = admin_pb2.AdminMessage()
@@ -250,7 +238,7 @@ class Node:
def setURL(self, url):
"""Set mesh network URL"""
if self.radioConfig == None:
if self.radioConfig is None:
raise Exception("No RadioConfig has been read")
# URLs are of the form https://www.meshtastic.org/d/#{base64_channel_set}
@@ -400,5 +388,3 @@ class Node:
wantResponse=wantResponse,
onResponse=onResponse,
channelIndex=adminIndex)

4
meshtastic/remote_hardware.py
View File

@@ -1,6 +1,6 @@
from . import portnums_pb2, remote_hardware_pb2
from pubsub import pub
from . import portnums_pb2, remote_hardware_pb2
def onGPIOreceive(packet, interface):
@@ -13,7 +13,7 @@ def onGPIOreceive(packet, interface):
class RemoteHardwareClient:
"""
This is the client code to control/monitor simple hardware built into the
This is the client code to control/monitor simple hardware built into the
meshtastic devices. It is intended to be both a useful API/service and example
code for how you can connect to your own custom meshtastic services
"""

23
meshtastic/test.py
View File

@@ -1,11 +1,13 @@
""" Testing
"""
import logging
from . import util
from . import SerialInterface, TCPInterface, BROADCAST_NUM
from pubsub import pub
import time
import sys
import threading, traceback
import traceback
from dotmap import DotMap
from pubsub import pub
from . import util
from . import SerialInterface, TCPInterface, BROADCAST_NUM
"""The interfaces we are using for our tests"""
interfaces = None
@@ -31,7 +33,7 @@ def onReceive(packet, interface):
if p.decoded.portnum == "TEXT_MESSAGE_APP":
# We only care a about clear text packets
if receivedPackets != None:
if receivedPackets is not None:
receivedPackets.append(p)
@@ -75,18 +77,19 @@ def testSend(fromInterface, toInterface, isBroadcast=False, asBinary=False, want
else:
fromInterface.sendData((f"Binary {testNumber}").encode(
"utf-8"), toNode, wantAck=wantAck)
for sec in range(60): # max of 60 secs before we timeout
for _ in range(60): # max of 60 secs before we timeout
time.sleep(1)
if (len(receivedPackets) >= 1):
if len(receivedPackets) >= 1:
return True
return False # Failed to send
def runTests(numTests=50, wantAck=False, maxFailures=0):
"""Run the tests."""
logging.info(f"Running {numTests} tests with wantAck={wantAck}")
numFail = 0
numSuccess = 0
for i in range(numTests):
for _ in range(numTests):
global testNumber
testNumber = testNumber + 1
isBroadcast = True
@@ -116,6 +119,7 @@ def runTests(numTests=50, wantAck=False, maxFailures=0):
def testThread(numTests=50):
"""Test thread"""
logging.info("Found devices, starting tests...")
runTests(numTests, wantAck=True)
# Allow a few dropped packets
@@ -128,6 +132,7 @@ def onConnection(topic=pub.AUTO_TOPIC):
def openDebugLog(portName):
"""Open the debug log file"""
debugname = "log" + portName.replace("/", "_")
logging.info(f"Writing serial debugging to {debugname}")
return open(debugname, 'w+', buffering=1)
@@ -141,7 +146,7 @@ def testAll():
Exception: If not enough devices are found
"""
ports = util.findPorts()
if (len(ports) < 2):
if len(ports) < 2:
raise Exception("Must have at least two devices connected to USB")
pub.subscribe(onConnection, "meshtastic.connection")

8
meshtastic/tunnel.py
View File

@@ -1,4 +1,5 @@
# code for IP tunnel over a mesh
""" Code for IP tunnel over a mesh
# Note python-pytuntap was too buggy
# using pip3 install pytap2
# make sure to "sudo setcap cap_net_admin+eip /usr/bin/python3.8" so python can access tun device without being root
@@ -12,11 +13,12 @@
# ping -i 30 -W 30 10.115.64.152
# FIXME: use a more optimal MTU
"""
from . import portnums_pb2
from pubsub import pub
import logging
import threading
from pubsub import pub
from . import portnums_pb2
# A new non standard log level that is lower level than DEBUG
LOG_TRACE = 5

19
meshtastic/util.py
View File

@@ -1,9 +1,13 @@
from collections import defaultdict
import serial, traceback
import serial.tools.list_ports
""" Utility functions.
"""
import traceback
from queue import Queue
import threading, sys, time, logging
import sys
import time
import logging
import threading
import serial
import serial.tools.list_ports
"""Some devices such as a seger jlink we never want to accidentally open"""
blacklistVids = dict.fromkeys([0x1366])
@@ -16,6 +20,7 @@ def stripnl(s):
def fixme(message):
"""raise an exception for things that needs to be fixed"""
raise Exception(f"FIXME: {message}")
@@ -34,7 +39,7 @@ def findPorts():
list -- a list of device paths
"""
l = list(map(lambda port: port.device,
filter(lambda port: port.vid != None and port.vid not in blacklistVids,
filter(lambda port: port.vid is not None and port.vid not in blacklistVids,
serial.tools.list_ports.comports())))
l.sort()
return l
@@ -48,6 +53,7 @@ class dotdict(dict):
class Timeout:
"""Timeout class"""
def __init__(self, maxSecs=20):
self.expireTime = 0
self.sleepInterval = 0.1
@@ -77,6 +83,7 @@ class DeferredExecution():
self.thread.start()
def queueWork(self, runnable):
""" Queue up the work"""
self.queue.put(runnable)
def _run(self):