Merge pull request #298 from mkinney/master

refactor code to util; add duplicate check

Author: mkinney

examples/scan_for_devices.py

@@ -3,8 +3,7 @@
 """
 
 import sys
-from meshtastic.supported_device import get_unique_vendor_ids, active_ports_on_supported_devices
-from meshtastic.util import detect_supported_devices
+from meshtastic.util import detect_supported_devices, get_unique_vendor_ids, active_ports_on_supported_devices
 
 # simple arg check
 if len(sys.argv) != 1:

meshtastic/supported_device.py

@@ -2,10 +2,6 @@
     It is used for auto detection as to which device might be connected.
 """
 
-import platform
-import subprocess
-import re
-
 # Goal is to detect which device and port to use from the supported devices
 # without installing any libraries that are not currently in the python meshtastic library
 
@@ -91,106 +87,3 @@ def __init__(self, name, version=None, for_firmware=None, device_class="esp32",
                      heltec_v1, heltec_v2_0, heltec_v2_1,
                      meshtastic_diy_v1, techo_1, rak4631_5005, rak4631_19003,
                      rak11200]
-
-
-def get_unique_vendor_ids():
-    """Return a set of unique vendor ids"""
-    vids = set()
-    for d in supported_devices:
-        if d.usb_vendor_id_in_hex:
-            vids.add(d.usb_vendor_id_in_hex)
-    return vids
-
-def get_devices_with_vendor_id(vid):
-    """Return a set of unique devices with the vendor id"""
-    sd = set()
-    for d in supported_devices:
-        if d.usb_vendor_id_in_hex == vid:
-            sd.add(d)
-    return sd
-
-def active_ports_on_supported_devices(sds):
-    """Return a set of active ports based on the supplied supported devices"""
-    ports = set()
-    baseports = set()
-    system = platform.system()
-
-    # figure out what possible base ports there are
-    for d in sds:
-        if system == "Linux":
-            baseports.add(d.baseport_on_linux)
-        elif system == "Darwin":
-            baseports.add(d.baseport_on_mac)
-        elif system == "Windows":
-            baseports.add(d.baseport_on_windows)
-
-    for bp in baseports:
-        if system == "Linux":
-            # see if we have any devices (ignoring any stderr output)
-            command = f'ls -al /dev/{bp}* 2> /dev/null'
-            #print(f'command:{command}')
-            _, ls_output = subprocess.getstatusoutput(command)
-            #print(f'ls_output:{ls_output}')
-            # if we got output, there are ports
-            if len(ls_output) > 0:
-                #print('got output')
-                # for each line of output
-                lines = ls_output.split('\n')
-                #print(f'lines:{lines}')
-                for line in lines:
-                    parts = line.split(' ')
-                    #print(f'parts:{parts}')
-                    port = parts[-1]
-                    #print(f'port:{port}')
-                    ports.add(port)
-        elif system == "Darwin":
-            # see if we have any devices (ignoring any stderr output)
-            command = f'ls -al /dev/{bp}* 2> /dev/null'
-            #print(f'command:{command}')
-            _, ls_output = subprocess.getstatusoutput(command)
-            #print(f'ls_output:{ls_output}')
-            # if we got output, there are ports
-            if len(ls_output) > 0:
-                #print('got output')
-                # for each line of output
-                lines = ls_output.split('\n')
-                #print(f'lines:{lines}')
-                for line in lines:
-                    parts = line.split(' ')
-                    #print(f'parts:{parts}')
-                    port = parts[-1]
-                    #print(f'port:{port}')
-                    ports.add(port)
-        elif system == "Windows":
-            # for each device in supported devices found
-            for d in sds:
-                # find the port(s)
-                com_ports = detect_windows_port(d)
-                #print(f'com_ports:{com_ports}')
-                # add all ports
-                for com_port in com_ports:
-                    ports.add(com_port)
-    return ports
-
-
-def detect_windows_port(sd):
-    """detect if Windows port"""
-    ports = set()
-
-    if sd:
-        system = platform.system()
-
-        if system == "Windows":
-            command = ('powershell.exe "[Console]::OutputEncoding = [Text.UTF8Encoding]::UTF8;'
-                       'Get-PnpDevice -PresentOnly | Where-Object{ ($_.DeviceId -like ')
-            command += f"'*{sd.usb_vendor_id_in_hex.upper()}*'"
-            command += ')} | Format-List"'
-
-            #print(f'command:{command}')
-            _, sp_output = subprocess.getstatusoutput(command)
-            #print(f'sp_output:{sp_output}')
-            p = re.compile(r'\(COM(.*)\)')
-            for x in p.findall(sp_output):
-                #print(f'x:{x}')
-                ports.add(f'COM{x}')
-    return ports

meshtastic/tests/test_util.py

@@ -12,7 +12,9 @@
                              remove_keys_from_dict, Timeout, hexstr,
                              ipstr, readnet_u16, findPorts, convert_mac_addr,
                              snake_to_camel, camel_to_snake, eliminate_duplicate_port,
-                             is_windows11)
+                             is_windows11, active_ports_on_supported_devices)
+
+from meshtastic.supported_device import SupportedDevice
 
 
 @pytest.mark.unit
@@ -336,6 +338,8 @@ def test_eliminate_duplicate_port():
     assert eliminate_duplicate_port(['/dev/cu.usbserial-0001', '/dev/cu.SLAB_USBtoUART']) == ['/dev/cu.usbserial-0001']
     assert eliminate_duplicate_port(['/dev/cu.usbmodem11301', '/dev/cu.wchusbserial11301']) == ['/dev/cu.wchusbserial11301']
     assert eliminate_duplicate_port(['/dev/cu.wchusbserial11301', '/dev/cu.usbmodem11301']) == ['/dev/cu.wchusbserial11301']
+    assert eliminate_duplicate_port(['/dev/cu.usbmodem53230051441', '/dev/cu.wchusbserial53230051441']) == ['/dev/cu.wchusbserial53230051441']
+    assert eliminate_duplicate_port(['/dev/cu.wchusbserial53230051441', '/dev/cu.usbmodem53230051441']) == ['/dev/cu.wchusbserial53230051441']
 
 @patch('platform.version', return_value='10.0.22000.194')
 @patch('platform.release', return_value='10')
@@ -377,3 +381,78 @@ def test_is_windows11_false_win8_1(patched_platform, patched_release):
     assert is_windows11() is False
     patched_platform.assert_called()
     patched_release.assert_called()
+
+
+@pytest.mark.unit
+@patch('platform.system', return_value='Linux')
+def test_active_ports_on_supported_devices_empty(mock_platform):
+    """Test active_ports_on_supported_devices()"""
+    sds = set()
+    assert active_ports_on_supported_devices(sds) == set()
+    mock_platform.assert_called()
+
+
+@pytest.mark.unit
+@patch('subprocess.getstatusoutput')
+@patch('platform.system', return_value='Linux')
+def test_active_ports_on_supported_devices_linux(mock_platform, mock_sp):
+    """Test active_ports_on_supported_devices()"""
+    mock_sp.return_value = (None, 'crw-rw-rw-  1 root        wheel   0x9000000 Feb  8 22:22 /dev/ttyUSBfake')
+    fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1', baseport_on_linux='ttyUSB')
+    fake_supported_devices = [fake_device]
+    assert active_ports_on_supported_devices(fake_supported_devices) == {'/dev/ttyUSBfake'}
+    mock_platform.assert_called()
+    mock_sp.assert_called()
+
+
+@pytest.mark.unit
+@patch('subprocess.getstatusoutput')
+@patch('platform.system', return_value='Darwin')
+def test_active_ports_on_supported_devices_mac(mock_platform, mock_sp):
+    """Test active_ports_on_supported_devices()"""
+    mock_sp.return_value = (None, 'crw-rw-rw-  1 root        wheel   0x9000000 Feb  8 22:22 /dev/cu.usbserial-foo')
+    fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1', baseport_on_linux='cu.usbserial-')
+    fake_supported_devices = [fake_device]
+    assert active_ports_on_supported_devices(fake_supported_devices) == {'/dev/cu.usbserial-foo'}
+    mock_platform.assert_called()
+    mock_sp.assert_called()
+
+
+@pytest.mark.unit
+@patch('meshtastic.util.detect_windows_port', return_value={'COM2'})
+@patch('platform.system', return_value='Windows')
+def test_active_ports_on_supported_devices_win(mock_platform, mock_dwp):
+    """Test active_ports_on_supported_devices()"""
+    fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1')
+    fake_supported_devices = [fake_device]
+    assert active_ports_on_supported_devices(fake_supported_devices) == {'COM2'}
+    mock_platform.assert_called()
+    mock_dwp.assert_called()
+
+
+@pytest.mark.unit
+@patch('subprocess.getstatusoutput')
+@patch('platform.system', return_value='Darwin')
+def test_active_ports_on_supported_devices_mac_no_duplicates_check(mock_platform, mock_sp):
+    """Test active_ports_on_supported_devices()"""
+    mock_sp.return_value = (None, ('crw-rw-rw-  1 root  wheel  0x9000005 Mar  8 10:05 /dev/cu.usbmodem53230051441\n'
+                                   'crw-rw-rw-  1 root  wheel  0x9000003 Mar  8 10:06 /dev/cu.wchusbserial53230051441'))
+    fake_device = SupportedDevice(name='a', for_firmware='tbeam', baseport_on_mac='cu.usbmodem')
+    fake_supported_devices = [fake_device]
+    assert active_ports_on_supported_devices(fake_supported_devices, False) == {'/dev/cu.usbmodem53230051441', '/dev/cu.wchusbserial53230051441'}
+    mock_platform.assert_called()
+    mock_sp.assert_called()
+
+
+@pytest.mark.unit
+@patch('subprocess.getstatusoutput')
+@patch('platform.system', return_value='Darwin')
+def test_active_ports_on_supported_devices_mac_duplicates_check(mock_platform, mock_sp):
+    """Test active_ports_on_supported_devices()"""
+    mock_sp.return_value = (None, ('crw-rw-rw-  1 root  wheel  0x9000005 Mar  8 10:05 /dev/cu.usbmodem53230051441\n'
+                                   'crw-rw-rw-  1 root  wheel  0x9000003 Mar  8 10:06 /dev/cu.wchusbserial53230051441'))
+    fake_device = SupportedDevice(name='a', for_firmware='tbeam', baseport_on_mac='cu.usbmodem')
+    fake_supported_devices = [fake_device]
+    assert active_ports_on_supported_devices(fake_supported_devices, True) == {'/dev/cu.wchusbserial53230051441'}
+    mock_platform.assert_called()
+    mock_sp.assert_called()

meshtastic/util.py

@@ -14,7 +14,8 @@
 import serial
 import serial.tools.list_ports
 import pkg_resources
-from meshtastic.supported_device import get_unique_vendor_ids, get_devices_with_vendor_id
+
+from meshtastic.supported_device import supported_devices
 
 """Some devices such as a seger jlink we never want to accidentally open"""
 blacklistVids = dict.fromkeys([0x1366])
@@ -433,3 +434,112 @@ def is_windows11():
             except Exception as e:
                 print(f'problem detecting win11 e:{e}')
     return is_win11
+
+
+def get_unique_vendor_ids():
+    """Return a set of unique vendor ids"""
+    vids = set()
+    for d in supported_devices:
+        if d.usb_vendor_id_in_hex:
+            vids.add(d.usb_vendor_id_in_hex)
+    return vids
+
+
+def get_devices_with_vendor_id(vid):
+    """Return a set of unique devices with the vendor id"""
+    sd = set()
+    for d in supported_devices:
+        if d.usb_vendor_id_in_hex == vid:
+            sd.add(d)
+    return sd
+
+
+def active_ports_on_supported_devices(sds, eliminate_duplicates=False):
+    """Return a set of active ports based on the supplied supported devices"""
+    ports = set()
+    baseports = set()
+    system = platform.system()
+
+    # figure out what possible base ports there are
+    for d in sds:
+        if system == "Linux":
+            baseports.add(d.baseport_on_linux)
+        elif system == "Darwin":
+            baseports.add(d.baseport_on_mac)
+        elif system == "Windows":
+            baseports.add(d.baseport_on_windows)
+
+    for bp in baseports:
+        if system == "Linux":
+            # see if we have any devices (ignoring any stderr output)
+            command = f'ls -al /dev/{bp}* 2> /dev/null'
+            #print(f'command:{command}')
+            _, ls_output = subprocess.getstatusoutput(command)
+            #print(f'ls_output:{ls_output}')
+            # if we got output, there are ports
+            if len(ls_output) > 0:
+                #print('got output')
+                # for each line of output
+                lines = ls_output.split('\n')
+                #print(f'lines:{lines}')
+                for line in lines:
+                    parts = line.split(' ')
+                    #print(f'parts:{parts}')
+                    port = parts[-1]
+                    #print(f'port:{port}')
+                    ports.add(port)
+        elif system == "Darwin":
+            # see if we have any devices (ignoring any stderr output)
+            command = f'ls -al /dev/{bp}* 2> /dev/null'
+            #print(f'command:{command}')
+            _, ls_output = subprocess.getstatusoutput(command)
+            #print(f'ls_output:{ls_output}')
+            # if we got output, there are ports
+            if len(ls_output) > 0:
+                #print('got output')
+                # for each line of output
+                lines = ls_output.split('\n')
+                #print(f'lines:{lines}')
+                for line in lines:
+                    parts = line.split(' ')
+                    #print(f'parts:{parts}')
+                    port = parts[-1]
+                    #print(f'port:{port}')
+                    ports.add(port)
+        elif system == "Windows":
+            # for each device in supported devices found
+            for d in sds:
+                # find the port(s)
+                com_ports = detect_windows_port(d)
+                #print(f'com_ports:{com_ports}')
+                # add all ports
+                for com_port in com_ports:
+                    ports.add(com_port)
+    if eliminate_duplicates:
+        ports = eliminate_duplicate_port(list(ports))
+        ports.sort()
+        ports = set(ports)
+    return ports
+
+
+def detect_windows_port(sd):
+    """detect if Windows port"""
+    ports = set()
+
+    if sd:
+        system = platform.system()
+
+        if system == "Windows":
+            command = ('powershell.exe "[Console]::OutputEncoding = [Text.UTF8Encoding]::UTF8;'
+                       'Get-PnpDevice -PresentOnly | Where-Object{ ($_.DeviceId -like ')
+            command += f"'*{sd.usb_vendor_id_in_hex.upper()}*'"
+            command += ')} | Format-List"'
+
+            #print(f'command:{command}')
+            _, sp_output = subprocess.getstatusoutput(command)
+            #print(f'sp_output:{sp_output}')
+            p = re.compile(r'\(COM(.*)\)')
+            for x in p.findall(sp_output):
+                #print(f'x:{x}')
+                ports.add(f'COM{x}')
+    return ports