diff --git a/.gitignore b/.gitignore index d3166984..19cefb6f 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ dist log_* .eggs nanopb-0.4.4 +nanopb-0.4.5 .*swp .coverage *.py-E diff --git a/examples/scan_for_devices.py b/examples/scan_for_devices.py index 9971a931..8c4f021e 100644 --- a/examples/scan_for_devices.py +++ b/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: diff --git a/meshtastic/supported_device.py b/meshtastic/supported_device.py index 82c760e0..d3b36b2b 100755 --- a/meshtastic/supported_device.py +++ b/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 diff --git a/meshtastic/tests/test_util.py b/meshtastic/tests/test_util.py index b04fd924..ed010703 100644 --- a/meshtastic/tests/test_util.py +++ b/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 @@ -335,6 +337,8 @@ def test_eliminate_duplicate_port(): assert eliminate_duplicate_port(['/dev/cu.SLAB_USBtoUART', '/dev/cu.usbserial-0001']) == ['/dev/cu.usbserial-0001'] 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.usbmodem53230051441', '/dev/cu.wchusbserial53230051441']) == ['/dev/cu.wchusbserial53230051441'] + assert eliminate_duplicate_port(['/dev/cu.wchusbserial53230051441', '/dev/cu.usbmodem53230051441']) == ['/dev/cu.wchusbserial53230051441'] assert eliminate_duplicate_port(['/dev/cu.wchusbserial11301', '/dev/cu.usbmodem11301']) == ['/dev/cu.wchusbserial11301'] @patch('platform.version', return_value='10.0.22000.194') @@ -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() diff --git a/meshtastic/util.py b/meshtastic/util.py index 79cf36f5..f9f05e15 100644 --- a/meshtastic/util.py +++ b/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