Wls add types2

meshtastic/__init__.py

@@ -108,13 +108,13 @@ def onConnection(interface, topic=pub.AUTO_TOPIC): # called when we (re)connect
 LOCAL_ADDR = "^local"
 """A special ID that means the local node"""
 
-BROADCAST_NUM = 0xFFFFFFFF
+BROADCAST_NUM: int = 0xFFFFFFFF
 """if using 8 bit nodenums this will be shortened on the target"""
 
 BROADCAST_ADDR = "^all"
 """A special ID that means broadcast"""
 
-OUR_APP_VERSION = 20300
+OUR_APP_VERSION: int = 20300
 """The numeric buildnumber (shared with android apps) specifying the
  level of device code we are guaranteed to understand
 

meshtastic/__main__.py

@@ -11,7 +11,7 @@
 import platform
 import sys
 import time
-from typing import Optional
+from typing import List, Optional
 
 import pyqrcode # type: ignore[import-untyped]
 import yaml
@@ -42,7 +42,7 @@
 from meshtastic.protobuf import channel_pb2, config_pb2, portnums_pb2
 from meshtastic.version import get_active_version
 
-def onReceive(packet, interface):
+def onReceive(packet, interface) -> None:
  """Callback invoked when a packet arrives"""
  args = mt_config.args
  try:
@@ -73,7 +73,7 @@ def onReceive(packet, interface):
  print(f"Warning: There is no field {ex} in the packet.")
 
 
-def onConnection(interface, topic=pub.AUTO_TOPIC): # pylint: disable=W0613
+def onConnection(interface, topic=pub.AUTO_TOPIC) -> None: # pylint: disable=W0613
  """Callback invoked when we connect/disconnect from a radio"""
  print(f"Connection changed: {topic.getName()}")
 
@@ -85,7 +85,7 @@ def checkChannel(interface: MeshInterface, channelIndex: int) -> bool:
  return ch and ch.role != channel_pb2.Channel.Role.DISABLED
 
 
-def getPref(node, comp_name):
+def getPref(node, comp_name) -> bool:
  """Get a channel or preferences value"""
 
  name = splitCompoundName(comp_name)
@@ -100,11 +100,11 @@ def getPref(node, comp_name):
  # First validate the input
  localConfig = node.localConfig
  moduleConfig = node.moduleConfig
- found = False
+ found: bool = False
  for config in [localConfig, moduleConfig]:
  objDesc = config.DESCRIPTOR
  config_type = objDesc.fields_by_name.get(name[0])
- pref = False
+ pref = "" #FIXME - is this correct to leave as an empty string if not found?
  if config_type:
  pref = config_type.message_type.fields_by_name.get(snake_name)
  if pref or wholeField:
@@ -151,16 +151,16 @@ def getPref(node, comp_name):
  return True
 
 
-def splitCompoundName(comp_name):
+def splitCompoundName(comp_name: str) -> List[str]:
  """Split compound (dot separated) preference name into parts"""
- name = comp_name.split(".")
+ name: List[str] = comp_name.split(".")
  if len(name) < 2:
  name[0] = comp_name
  name.append(comp_name)
  return name
 
 
-def traverseConfig(config_root, config, interface_config):
+def traverseConfig(config_root, config, interface_config) -> bool:
  """Iterate through current config level preferences and either traverse deeper if preference is a dict or set preference"""
  snake_name = meshtastic.util.camel_to_snake(config_root)
  for pref in config:
@@ -955,7 +955,7 @@ def setSimpleConfig(modem_preset):
  sys.exit(1)
 
 
-def printConfig(config):
+def printConfig(config) -> None:
  """print configuration"""
  objDesc = config.DESCRIPTOR
  for config_section in objDesc.fields:
@@ -972,12 +972,12 @@ def printConfig(config):
  print(f" {temp_name}")
 
 
-def onNode(node):
+def onNode(node) -> None:
  """Callback invoked when the node DB changes"""
  print(f"Node changed: {node}")
 
 
-def subscribe():
+def subscribe() -> None:
  """Subscribe to the topics the user probably wants to see, prints output to stdout"""
  pub.subscribe(onReceive, "meshtastic.receive")
  # pub.subscribe(onConnection, "meshtastic.connection")
@@ -988,7 +988,7 @@ def subscribe():
  # pub.subscribe(onNode, "meshtastic.node")
 
 
-def export_config(interface):
+def export_config(interface) -> str:
  """used in --export-config"""
  configObj = {}
 
@@ -1020,7 +1020,7 @@ def export_config(interface):
  if alt:
  configObj["location"]["alt"] = alt
 
- config = MessageToDict(interface.localNode.localConfig)
+ config = MessageToDict(interface.localNode.localConfig) #checkme - Used as a dictionary here and a string below
  if config:
  # Convert inner keys to correct snake/camelCase
  prefs = {}
@@ -1030,7 +1030,7 @@ def export_config(interface):
  else:
  prefs[pref] = config[pref]
  if mt_config.camel_case:
- configObj["config"] = config
+ configObj["config"] = config #Identical command here and 2 lines below?
  else:
  configObj["config"] = config
 
@@ -1046,10 +1046,11 @@ def export_config(interface):
  else:
  configObj["module_config"] = prefs
 
- config = "# start of Meshtastic configure yaml\n"
- config += yaml.dump(configObj)
- print(config)
- return config
+ config_txt = "# start of Meshtastic configure yaml\n" #checkme - "config" (now changed to config_out)
+ #was used as a string here and a Dictionary above
+ config_txt += yaml.dump(configObj)
+ print(config_txt)
+ return config_txt
 
 
 def create_power_meter():

meshtastic/ble_interface.py

@@ -5,6 +5,7 @@
 import logging
 import struct
 import time
+import io
 from threading import Thread
 from typing import List, Optional
 
@@ -34,9 +35,9 @@ def __init__(
  self,
  address: Optional[str],
  noProto: bool = False,
- debugOut=None,
+ debugOut: Optional[io.TextIOWrapper]=None,
  noNodes: bool = False,
- ):
+ ) -> None:
  MeshInterface.__init__(
  self, debugOut=debugOut, noProto=noProto, noNodes=noNodes
  )
@@ -82,7 +83,7 @@ def __init__(
  # Note: the on disconnected callback will call our self.close which will make us nicely wait for threads to exit
  self._exit_handler = atexit.register(self.client.disconnect)
 
- def from_num_handler(self, _, b): # pylint: disable=C0116
+ def from_num_handler(self, _, b: bytes) -> None: # pylint: disable=C0116
  """Handle callbacks for fromnum notify.
  Note: this method does not need to be async because it is just setting a bool.
  """
@@ -150,9 +151,12 @@ def find_device(self, address: Optional[str]) -> BLEDevice:
  )
  return addressed_devices[0]
 
- def _sanitize_address(address): # pylint: disable=E0213
+ def _sanitize_address(self, address: Optional[str]) -> Optional[str]: # pylint: disable=E0213
  "Standardize BLE address by removing extraneous characters and lowercasing."
- return address.replace("-", "").replace("_", "").replace(":", "").lower()
+ if address is None:
+ return None
+ else:
+ return address.replace("-", "").replace("_", "").replace(":", "").lower()
 
  def connect(self, address: Optional[str] = None) -> "BLEClient":
  "Connect to a device by address."
@@ -164,11 +168,11 @@ def connect(self, address: Optional[str] = None) -> "BLEClient":
  client.discover()
  return client
 
- def _receiveFromRadioImpl(self):
+ def _receiveFromRadioImpl(self) -> None:
  while self._want_receive:
  if self.should_read:
  self.should_read = False
- retries = 0
+ retries: int = 0
  while self._want_receive:
  try:
  b = bytes(self.client.read_gatt_char(FROMRADIO_UUID))
@@ -194,8 +198,8 @@ def _receiveFromRadioImpl(self):
  else:
  time.sleep(0.01)
 
- def _sendToRadioImpl(self, toRadio):
- b = toRadio.SerializeToString()
+ def _sendToRadioImpl(self, toRadio) -> None:
+ b: bytes = toRadio.SerializeToString()
  if b and self.client: # we silently ignore writes while we are shutting down
  logging.debug(f"TORADIO write: {b.hex()}")
  try:
@@ -211,7 +215,7 @@ def _sendToRadioImpl(self, toRadio):
  time.sleep(0.01)
  self.should_read = True
 
- def close(self):
+ def close(self) -> None:
  try:
  MeshInterface.close(self)
  except Exception as e:
@@ -235,7 +239,7 @@ def close(self):
 class BLEClient:
  """Client for managing connection to a BLE device"""
 
- def __init__(self, address=None, **kwargs):
+ def __init__(self, address=None, **kwargs) -> None:
  self._eventLoop = asyncio.new_event_loop()
  self._eventThread = Thread(
  target=self._run_event_loop, name="BLEClient", daemon=True

meshtastic/remote_hardware.py

@@ -8,7 +8,7 @@
 from meshtastic.util import our_exit
 
 
-def onGPIOreceive(packet, interface):
+def onGPIOreceive(packet, interface) -> None:
  """Callback for received GPIO responses"""
  logging.debug(f"packet:{packet} interface:{interface}")
  gpioValue = 0
@@ -37,7 +37,7 @@ class RemoteHardwareClient:
  code for how you can connect to your own custom meshtastic services
  """
 
- def __init__(self, iface):
+ def __init__(self, iface) -> None:
  """
  Constructor
 

meshtastic/serial_interface.py

@@ -4,7 +4,7 @@
 import platform
 import time
 
-from typing import Optional
+from typing import List, Optional
 
 import serial # type: ignore[import-untyped]
 
@@ -18,7 +18,7 @@
 class SerialInterface(StreamInterface):
  """Interface class for meshtastic devices over a serial link"""
 
- def __init__(self, devPath: Optional[str]=None, debugOut=None, noProto=False, connectNow=True, noNodes: bool=False):
+ def __init__(self, devPath: Optional[str]=None, debugOut=None, noProto: bool=False, connectNow: bool=True, noNodes: bool=False) -> None:
  """Constructor, opens a connection to a specified serial port, or if unspecified try to
  find one Meshtastic device by probing
 
@@ -31,13 +31,13 @@ def __init__(self, devPath: Optional[str]=None, debugOut=None, noProto=False, co
  self.devPath: Optional[str] = devPath
 
  if self.devPath is None:
- ports = meshtastic.util.findPorts(True)
+ ports: List[str] = meshtastic.util.findPorts(True)
  logging.debug(f"ports:{ports}")
  if len(ports) == 0:
  print("No Serial Meshtastic device detected, attempting TCP connection on localhost.")
  return
  elif len(ports) > 1:
- message = "Warning: Multiple serial ports were detected so one serial port must be specified with the '--port'.\n"
+ message: str = "Warning: Multiple serial ports were detected so one serial port must be specified with the '--port'.\n"
  message += f" Ports detected:{ports}"
  meshtastic.util.our_exit(message)
  else:
@@ -58,14 +58,14 @@ def __init__(self, devPath: Optional[str]=None, debugOut=None, noProto=False, co
  self.stream = serial.Serial(
  self.devPath, 115200, exclusive=True, timeout=0.5, write_timeout=0
  )
- self.stream.flush()
+ self.stream.flush() # type: ignore[attr-defined]
  time.sleep(0.1)
 
  StreamInterface.__init__(
  self, debugOut=debugOut, noProto=noProto, connectNow=connectNow, noNodes=noNodes
  )
 
- def close(self):
+ def close(self) -> None:
  """Close a connection to the device"""
  if self.stream: # Stream can be null if we were already closed
  self.stream.flush() # FIXME: why are there these two flushes with 100ms sleeps? This shouldn't be necessary