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 #517 from ianmcorvidae/pylint-wrangling

Browse Source 
Pylint wrangling
This commit is contained in:
Ben Meadors
2024-03-19 17:51:14 -05:00
committed by GitHub
parent 0528a6fb3b 9b5943192d
commit 341d8e0cec
12 changed files with 73 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
meshtastic/__main__.py
View File

@@ -1260,7 +1260,7 @@ def initParser():
help="Request telemetry from a node. "
"You need pass the destination ID as argument with '--dest'. "
"For repeaters, the nodeNum is required.",
action="store_true",
action="store_true",
)
parser.add_argument(

55
meshtastic/ble_interface.py
View File

@@ -3,12 +3,12 @@
import logging
import time
import struct
import asyncio
from threading import Thread, Event
from bleak import BleakScanner, BleakClient
from meshtastic.mesh_interface import MeshInterface
from meshtastic.util import our_exit
from bleak import BleakScanner, BleakClient
import asyncio
SERVICE_UUID = "6ba1b218-15a8-461f-9fa8-5dcae273eafd"
TORADIO_UUID = "f75c76d2-129e-4dad-a1dd-7866124401e7"
@@ -17,13 +17,14 @@ FROMNUM_UUID = "ed9da18c-a800-4f66-a670-aa7547e34453"
class BLEInterface(MeshInterface):
"""MeshInterface using BLE to connect to devices"""
class BLEError(Exception):
"""An exception class for BLE errors"""
def __init__(self, message):
self.message = message
super().__init__(self.message)
class BLEState():
class BLEState(): # pylint: disable=C0115
THREADS = False
BLE = False
MESH = False
@@ -69,13 +70,14 @@ class BLEInterface(MeshInterface):
self.client.start_notify(FROMNUM_UUID, self.from_num_handler)
async def from_num_handler(self, _, b):
async def from_num_handler(self, _, b): # pylint: disable=C0116
from_num = struct.unpack('<I', bytes(b))[0]
logging.debug(f"FROMNUM notify: {from_num}")
self.should_read = True
def scan(self):
"Scan for available BLE devices"
with BLEClient() as client:
return [
(x[0], x[1]) for x in (client.discover(
@@ -86,12 +88,15 @@ class BLEInterface(MeshInterface):
def find_device(self, address):
"Find a device by address"
meshtastic_devices = self.scan()
addressed_devices = list(filter(lambda x: address == x[1].local_name or address == x[0].name, meshtastic_devices))
addressed_devices = list(filter(lambda x: address in (x[1].local_name, x[0].name), meshtastic_devices))
# If nothing is found try on the address
if len(addressed_devices) == 0:
addressed_devices = list(filter(lambda x: BLEInterface._sanitize_address(address) == BLEInterface._sanitize_address(x[0].address), meshtastic_devices))
addressed_devices = list(filter(
lambda x: BLEInterface._sanitize_address(address) == BLEInterface._sanitize_address(x[0].address),
meshtastic_devices))
if len(addressed_devices) == 0:
raise BLEInterface.BLEError(f"No Meshtastic BLE peripheral with identifier or address '{address}' found. Try --ble-scan to find it.")
@@ -99,7 +104,8 @@ class BLEInterface(MeshInterface):
raise BLEInterface.BLEError(f"More than one Meshtastic BLE peripheral with identifier or address '{address}' found.")
return addressed_devices[0][0]
def _sanitize_address(address):
def _sanitize_address(address): # pylint: disable=E0213
"Standardize BLE address by removing extraneous characters and lowercasing"
return address \
.replace("-", "") \
.replace("_", "") \
@@ -107,6 +113,7 @@ class BLEInterface(MeshInterface):
.lower()
def connect(self, address):
"Connect to a device by address"
device = self.find_device(address)
client = BLEClient(device.address)
client.connect()
@@ -156,13 +163,14 @@ class BLEInterface(MeshInterface):
if self.state.THREADS:
self._receiveThread_started.clear()
self._receiveThread_stopped.wait(5)
if self.state.BLE:
self.client.disconnect()
self.client.close()
class BLEClient():
"""Client for managing connection to a BLE device"""
def __init__(self, address = None, **kwargs):
self._eventThread = Thread(target = self._run_event_loop)
self._eventThread_started = Event()
@@ -177,47 +185,46 @@ class BLEClient():
self.bleak_client = BleakClient(address, **kwargs)
def discover(self, **kwargs):
def discover(self, **kwargs): # pylint: disable=C0116
return self.async_await(BleakScanner.discover(**kwargs))
def pair(self, **kwargs):
def pair(self, **kwargs): # pylint: disable=C0116
return self.async_await(self.bleak_client.pair(**kwargs))
def connect(self, **kwargs):
def connect(self, **kwargs): # pylint: disable=C0116
return self.async_await(self.bleak_client.connect(**kwargs))
def disconnect(self, **kwargs):
def disconnect(self, **kwargs): # pylint: disable=C0116
self.async_await(self.bleak_client.disconnect(**kwargs))
def read_gatt_char(self, *args, **kwargs):
def read_gatt_char(self, *args, **kwargs): # pylint: disable=C0116
return self.async_await(self.bleak_client.read_gatt_char(*args, **kwargs))
def write_gatt_char(self, *args, **kwargs):
def write_gatt_char(self, *args, **kwargs): # pylint: disable=C0116
self.async_await(self.bleak_client.write_gatt_char(*args, **kwargs))
def start_notify(self, *args, **kwargs):
def start_notify(self, *args, **kwargs): # pylint: disable=C0116
self.async_await(self.bleak_client.start_notify(*args, **kwargs))
def close(self):
def close(self): # pylint: disable=C0116
self.async_run(self._stop_event_loop())
self._eventThread_stopped.wait(5)
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
def __exit__(self, _type, _value, _traceback):
self.close()
def async_await(self, coro, timeout = None):
def async_await(self, coro, timeout = None): # pylint: disable=C0116
return self.async_run(coro).result(timeout)
def async_run(self, coro):
def async_run(self, coro): # pylint: disable=C0116
return asyncio.run_coroutine_threadsafe(coro, self._eventLoop)
def _run_event_loop(self):
self._eventLoop = asyncio.new_event_loop()
# I don't know if the event loop can be initialized in __init__ so silencing pylint
self._eventLoop = asyncio.new_event_loop() # pylint: disable=W0201
self._eventThread_started.set()
try:
self._eventLoop.run_forever()

2
meshtastic/globals.py
View File

@@ -24,7 +24,7 @@ class Globals:
def __init__(self):
"""Constructor for the Globals CLass"""
if Globals.__instance is not None:
raise Exception("This class is a singleton")
raise Exception("This class is a singleton") # pylint: disable=W0719
else:
Globals.__instance = self
self.args = None

31
meshtastic/mesh_interface.py
View File

@@ -23,7 +23,6 @@ from meshtastic.__init__ import (
BROADCAST_ADDR,
BROADCAST_NUM,
LOCAL_ADDR,
OUR_APP_VERSION,
ResponseHandler,
protocols,
publishingThread,
@@ -48,6 +47,12 @@ class MeshInterface:
debugOut
"""
class MeshInterfaceError(Exception):
"""An exception class for general mesh interface errors"""
def __init__(self, message):
self.message = message
super().__init__(self.message)
def __init__(self, debugOut=None, noProto=False):
"""Constructor
@@ -314,7 +319,7 @@ class MeshInterface:
f"mesh_pb2.Constants.DATA_PAYLOAD_LEN: {mesh_pb2.Constants.DATA_PAYLOAD_LEN}"
)
if len(data) > mesh_pb2.Constants.DATA_PAYLOAD_LEN:
raise Exception("Data payload too big")
raise MeshInterface.MeshInterfaceError("Data payload too big")
if (
portNum == portnums_pb2.PortNum.UNKNOWN_APP
@@ -440,7 +445,7 @@ class MeshInterface:
destinationId = int(destinationId[1:], 16)
else:
destinationId = int(destinationId)
self.sendData(
r,
destinationId=destinationId,
@@ -469,7 +474,7 @@ class MeshInterface:
)
if telemetry.device_metrics.air_util_tx is not None:
print(f"Transmit air utilization: {telemetry.device_metrics.air_util_tx:.2f}%")
elif p["decoded"]["portnum"] == 'ROUTING_APP':
if p["decoded"]["routing"]["errorReason"] == 'NO_RESPONSE':
our_exit("No response from node. At least firmware 2.1.22 is required on the destination node.")
@@ -543,25 +548,25 @@ class MeshInterface:
and self.localNode.waitForConfig()
)
if not success:
raise Exception("Timed out waiting for interface config")
raise MeshInterface.MeshInterfaceError("Timed out waiting for interface config")
def waitForAckNak(self):
"""Wait for the ack/nak"""
success = self._timeout.waitForAckNak(self._acknowledgment)
if not success:
raise Exception("Timed out waiting for an acknowledgment")
raise MeshInterface.MeshInterfaceError("Timed out waiting for an acknowledgment")
def waitForTraceRoute(self, waitFactor):
"""Wait for trace route"""
success = self._timeout.waitForTraceRoute(waitFactor, self._acknowledgment)
if not success:
raise Exception("Timed out waiting for traceroute")
raise MeshInterface.MeshInterfaceError("Timed out waiting for traceroute")
def waitForTelemetry(self):
"""Wait for telemetry"""
success = self._timeout.waitForTelemetry(self._acknowledgment)
if not success:
raise Exception("Timed out waiting for telemetry")
raise MeshInterface.MeshInterfaceError("Timed out waiting for telemetry")
def getMyNodeInfo(self):
"""Get info about my node."""
@@ -596,7 +601,7 @@ class MeshInterface:
and raise an exception"""
if not self.noProto:
if not self.isConnected.wait(timeout): # timeout after x seconds
raise Exception("Timed out waiting for connection completion")
raise MeshInterface.MeshInterfaceError("Timed out waiting for connection completion")
# If we failed while connecting, raise the connection to the client
if self.failure:
@@ -605,7 +610,7 @@ class MeshInterface:
def _generatePacketId(self):
"""Get a new unique packet ID"""
if self.currentPacketId is None:
raise Exception("Not connected yet, can not generate packet")
raise MeshInterface.MeshInterfaceError("Not connected yet, can not generate packet")
else:
self.currentPacketId = (self.currentPacketId + 1) & 0xFFFFFFFF
return self.currentPacketId
@@ -774,7 +779,7 @@ class MeshInterface:
failmsg = None
if failmsg:
self.failure = Exception(failmsg)
self.failure = MeshInterface.MeshInterfaceError(failmsg)
self.isConnected.set() # let waitConnected return this exception
self.close()
@@ -920,7 +925,7 @@ class MeshInterface:
def _getOrCreateByNum(self, nodeNum):
"""Given a nodenum find the NodeInfo in the DB (or create if necessary)"""
if nodeNum == BROADCAST_NUM:
raise Exception("Can not create/find nodenum by the broadcast num")
raise MeshInterface.MeshInterfaceError("Can not create/find nodenum by the broadcast num")
if nodeNum in self.nodesByNum:
return self.nodesByNum[nodeNum]

5
meshtastic/node.py
View File

@@ -115,6 +115,7 @@ class Node:
print(f"{str(camel_to_snake(field))}:\n{str(config_values)}")
def requestConfig(self, configType):
"""Request the config from the node via admin message"""
if self == self.iface.localNode:
onResponse = None
else:
@@ -688,9 +689,6 @@ class Node:
logging.debug(f"Received channel {stripnl(c)}")
index = c.index
# for stress testing, we can always download all channels
fastChannelDownload = True
if index >= 8 - 1:
logging.debug("Finished downloading channels")
@@ -703,6 +701,7 @@ class Node:
self._requestChannel(index + 1)
def onAckNak(self, p):
"""Informative handler for ACK/NAK responses"""
if p["decoded"]["routing"]["errorReason"] != "NONE":
print(
f'Received a NAK, error reason: {p["decoded"]["routing"]["errorReason"]}'

2
meshtastic/stream_interface.py
View File

@@ -32,7 +32,7 @@ class StreamInterface(MeshInterface):
"""
if not hasattr(self, "stream") and not noProto:
raise Exception(
raise Exception( # pylint: disable=W0719
"StreamInterface is now abstract (to update existing code create SerialInterface instead)"
)
self._rxBuf = bytes() # empty

4
meshtastic/tests/test_main.py
View File

@@ -23,7 +23,7 @@ from meshtastic.__main__ import (
tunnelMain,
)
from ..channel_pb2 import Channel
from ..channel_pb2 import Channel # pylint: disable=E0611
# from ..ble_interface import BLEInterface
from ..node import Node
@@ -388,7 +388,7 @@ def test_main_onConnected_exception(capsys):
Globals.getInstance().set_args(sys.argv)
def throw_an_exception(junk):
raise Exception("Fake exception.")
raise Exception("Fake exception.") # pylint: disable=W0719
iface = MagicMock(autospec=SerialInterface)
with patch("meshtastic.serial_interface.SerialInterface", return_value=iface):

2
meshtastic/tests/test_node.py
View File

@@ -7,7 +7,7 @@ from unittest.mock import MagicMock, patch
import pytest
# from ..admin_pb2 import AdminMessage
from ..channel_pb2 import Channel
from ..channel_pb2 import Channel # pylint: disable=E0611
from ..node import Node
from ..serial_interface import SerialInterface

2
meshtastic/tests/test_util.py
View File

@@ -177,7 +177,7 @@ def test_catchAndIgnore(caplog):
"""Test catchAndIgnore() does not actually throw an exception, but just logs"""
def some_closure():
raise Exception("foo")
raise Exception("foo") # pylint: disable=W0719
with caplog.at_level(logging.DEBUG):
catchAndIgnore("something", some_closure)

14
meshtastic/tunnel.py
View File

@@ -38,6 +38,12 @@ def onTunnelReceive(packet, interface): # pylint: disable=W0613
class Tunnel:
"""A TUN based IP tunnel over meshtastic"""
class TunnelError(Exception):
"""An exception class for general tunnel errors"""
def __init__(self, message):
self.message = message
super().__init__(self.message)
def __init__(self, iface, subnet="10.115", netmask="255.255.0.0"):
"""
Constructor
@@ -47,19 +53,19 @@ class Tunnel:
"""
if not iface:
raise Exception("Tunnel() must have a interface")
raise Tunnel.TunnelError("Tunnel() must have a interface")
if not subnet:
raise Exception("Tunnel() must have a subnet")
raise Tunnel.TunnelError("Tunnel() must have a subnet")
if not netmask:
raise Exception("Tunnel() must have a netmask")
raise Tunnel.TunnelError("Tunnel() must have a netmask")
self.iface = iface
self.subnetPrefix = subnet
if platform.system() != "Linux":
raise Exception("Tunnel() can only be run instantiated on a Linux system")
raise Tunnel.TunnelError("Tunnel() can only be run instantiated on a Linux system")
our_globals = Globals.getInstance()
our_globals.set_tunnelInstance(self)

4
meshtastic/util.py
View File

@@ -109,7 +109,7 @@ def stripnl(s):
def fixme(message):
"""Raise an exception for things that needs to be fixed"""
raise Exception(f"FIXME: {message}")
raise Exception(f"FIXME: {message}") # pylint: disable=W0719
def catchAndIgnore(reason, closure):
@@ -193,7 +193,7 @@ class Timeout:
return True
time.sleep(self.sleepInterval)
return False
def waitForTelemetry(self, acknowledgment):
"""Block until telemetry response is received. Returns True if telemetry response has been received."""
self.reset()

4
meshtastic/version.py
View File

@@ -1,3 +1,4 @@
"""Version lookup utilities, isolated for cleanliness"""
import sys
try:
from importlib.metadata import version
@@ -5,7 +6,8 @@ except:
import pkg_resources
def get_active_version():
"""Get the currently active version using importlib, or pkg_resources if we must"""
if "importlib.metadata" in sys.modules:
return version("meshtastic")
else:
return pkg_resources.get_distribution("meshtastic").version
return pkg_resources.get_distribution("meshtastic").version # pylint: disable=E0601