import os import shutil import sys import time from logging import getLogger from textwrap import dedent from tkinter import messagebox, ttk from typing import Any, Dict, List, Optional, Tuple from thonny import get_runner, get_shell, get_workbench, running, ui_utils from thonny.common import CommandToBackend, EOFCommand, ImmediateCommand, InlineCommand from thonny.languages import tr from thonny.misc_utils import list_volumes from thonny.plugins.backend_config_page import ( BackendDetailsConfigPage, BaseSshProxyConfigPage, get_ssh_password, ) from thonny.running import SubprocessProxy from thonny.ui_utils import create_string_var, create_url_label, ems_to_pixels logger = getLogger(__name__) DEFAULT_WEBREPL_URL = "ws://192.168.4.1:8266/" WEBREPL_PORT_VALUE = "webrepl" VIDS_PIDS_TO_AVOID_IN_GENERIC_BACKEND = set() class MicroPythonProxy(SubprocessProxy): def __init__(self, clean): self._lib_dirs = [] super().__init__(clean, running.get_front_interpreter_for_subprocess()) def get_pip_gui_class(self): return None def get_pip_target_dir(self) -> Optional[str]: lib_dirs = self.get_lib_dirs() if not lib_dirs: return None for path in lib_dirs: if path.startswith("/home/"): return path for path in ["/lib", "/flash/lib"]: if path in lib_dirs: return path return lib_dirs[0] def get_lib_dirs(self): return self._lib_dirs def _store_state_info(self, msg): super(MicroPythonProxy, self)._store_state_info(msg) if "lib_dirs" in msg: self._lib_dirs = msg["lib_dirs"] def _get_time_args(self): result = { "sync_time": get_workbench().get_option(self.backend_name + ".sync_time", False), "validate_time": get_workbench().get_option( self.backend_name + ".validate_time", False ), "local_rtc": get_workbench().get_option(self.backend_name + ".local_rtc", False), } return result def _installer_runs_locally(self): return True def has_local_interpreter(self): return False def can_debug(self) -> bool: return False def can_run_in_terminal(self) -> bool: return False @classmethod def is_valid_configuration(cls, conf: Dict[str, Any]) -> bool: return True class BareMetalMicroPythonProxy(MicroPythonProxy): def __init__(self, clean): self._port = get_workbench().get_option(self.backend_name + ".port") self._clean_start = clean self._fix_port() super().__init__(clean) def get_target_executable(self) -> Optional[str]: return None def get_pip_gui_class(self): from thonny.plugins.micropython.pip_gui import MicroPythonPipDialog return MicroPythonPipDialog def destroy(self, for_restart: bool = False): super().destroy(for_restart=for_restart) if self._port != WEBREPL_PORT_VALUE: # let the OS release the port time.sleep(0.1) def _fix_port(self): if self._port == WEBREPL_PORT_VALUE: return elif self._port == "auto": potential = self._detect_potential_ports() if len(potential) == 1: self._port = potential[0][0] else: self._port = None message = dedent( """\ Couldn't find the device automatically. Check the connection (making sure the device is not in bootloader mode) or choose "Configure interpreter" in the interpreter menu (bottom-right corner of the window) to select specific port or another interpreter.""" ) if len(potential) > 1: _, descriptions = zip(*potential) message += "\n\nLikely candidates are:\n * " + "\n * ".join(descriptions) self._show_error(message) def _start_background_process(self, clean=None, extra_args=[]): if self._port is None: return super()._start_background_process(clean=clean, extra_args=extra_args) def _get_launcher_with_args(self): args = { "clean": self._clean_start, "port": self._port, "dtr": get_workbench().get_option(self.backend_name + ".dtr"), "rts": get_workbench().get_option(self.backend_name + ".rts"), "submit_mode": get_workbench().get_option(self.backend_name + ".submit_mode"), "interrupt_on_connect": get_workbench().get_option( self.backend_name + ".interrupt_on_connect" ), "write_block_size": self._get_write_block_size(), "write_block_delay": self._get_write_block_delay(), "proxy_class": self.__class__.__name__, } if self._port == WEBREPL_PORT_VALUE: args["url"] = get_workbench().get_option(self.backend_name + ".webrepl_url") args["password"] = get_workbench().get_option(self.backend_name + ".webrepl_password") args.update(self._get_time_args()) cmd = [ self._get_backend_launcher_path(), repr(args), ] return cmd def should_restart_interpreter_before_run(self): return get_workbench().get_option(self.backend_name + ".restart_interpreter_before_run") def stop_restart_kills_user_program(self) -> bool: return False def _get_backend_launcher_path(self) -> str: import thonny.plugins.micropython.bare_metal_backend return thonny.plugins.micropython.bare_metal_backend.__file__ def _get_write_block_size(self): return get_workbench().get_option(self.backend_name + ".write_block_size") def _get_write_block_delay(self): return get_workbench().get_option(self.backend_name + ".write_block_delay") def interrupt(self): # Don't interrupt local process, but direct it to device self._send_msg(ImmediateCommand("interrupt")) def send_command(self, cmd: CommandToBackend) -> Optional[str]: if isinstance(cmd, EOFCommand): # Runner doesn't notice restart get_shell().restart(was_running=get_runner().is_running()) return super().send_command(cmd) def _prepare_clean_launch(self): """Nothing to do in this level. The backend takes care of the clearing""" @classmethod def _detect_potential_ports(cls) -> List[Tuple[str, str]]: all_ports = list_serial_ports() """ for p in all_ports: print(vars(p)) """ last_backs = {} # get_workbench().get_option("serial.last_backend_per_vid_pid") return [(p.device, p.description) for p in all_ports if cls._is_potential_port(p)] @classmethod def _is_for_micropython(cls): return True @classmethod def _is_for_circuitpython(cls): return False @classmethod def _is_potential_port(cls, p): if "CircuitPython CDC2 " in (p.interface or ""): return False last_backs = get_workbench().get_option("serial.last_backend_per_vid_pid") if last_backs.get((p.vid, p.pid), "") == cls.backend_name: return True if "CircuitPython CDC " in (p.interface or ""): return cls._is_for_circuitpython() return ( (p.vid, p.pid) in cls.get_known_usb_vids_pids() or (p.vid, None) in cls.get_known_usb_vids_pids() or p.description in cls.get_known_port_descriptions() or cls.should_consider_unknown_devices() and (p.vid, p.pid) not in cls.get_vids_pids_to_avoid() and ( ("USB" in p.description and "serial" in p.description.lower()) or "UART" in p.description or "DAPLink" in p.description or "STLink" in p.description or "python" in p.description.lower() or "MicroPython" in (p.manufacturer or "") ) ) @classmethod def get_known_usb_vids_pids(cls): """Return set of pairs of USB device VID, PID""" return set() @classmethod def get_vids_pids_to_avoid(cls): """Return set of pairs of USB device VID, PID to explicitly not consider either because they are not compatible or to reduce the number of choices in the switcher. """ return set() @classmethod def should_consider_unknown_devices(cls): return True @classmethod def get_known_port_descriptions(cls): return set() def supports_remote_files(self): return self.is_connected() def uses_local_filesystem(self): return False def ready_for_remote_file_operations(self): return self.is_connected() and get_runner().is_waiting_toplevel_command() def supports_remote_directories(self): return self._cwd is not None and self._cwd != "" def supports_trash(self): return False def is_connected(self): return self._port is not None and self._proc is not None def _show_error(self, text): get_shell().print_error("\n" + text + "\n") def disconnect(self): self.destroy() def get_node_label(self): if "CircuitPython" in self._welcome_text: return tr("CircuitPython device") elif "micro:bit" in self._welcome_text.lower(): return "micro:bit" else: return tr("MicroPython device") def get_full_label(self): if self.is_connected(): return self.get_node_label() + " @ " + self._port else: return self.get_node_label() + " (" + tr("Not connected") + ")" def get_exe_dirs(self): return [] def can_run_local_files(self): return False def can_run_remote_files(self): return False def _check_remember_current_configuration(self) -> None: super()._check_remember_current_configuration() current_conf = self.get_current_switcher_configuration() port = current_conf[f"{self.backend_name}.port"] if port and port != WEBREPL_PORT_VALUE: last_backs = get_workbench().get_option("serial.last_backend_per_vid_pid") info = get_port_info(port) last_backs[(info.vid, info.pid)] = self.backend_name get_workbench().set_option("serial.last_backend_per_vid_pid", last_backs) def _should_remember_configuration(self, configuration: Dict[str, Any]) -> bool: return bool(configuration.get(f"{self.backend_name}.webrepl_url", False)) def get_current_switcher_configuration(self) -> Dict[str, Any]: # NB! using current port value, not the configured one (which may be "auto") conf = { "run.backend_name": self.backend_name, f"{self.backend_name}.port": self._port, } if self._port == WEBREPL_PORT_VALUE: conf[f"{self.backend_name}.webrepl_url"] = get_workbench().get_option( f"{self.backend_name}.webrepl_url" ) return conf @classmethod def get_switcher_configuration_label(cls, conf: Dict[str, Any]) -> str: port = conf[f"{cls.backend_name}.port"] if port == WEBREPL_PORT_VALUE: url = conf[f"{cls.backend_name}.webrepl_url"] return f"{cls.backend_description} • {url}" else: try: p = get_port_info(port) except Exception: p = None logger.exception("Could not get port info for %r", port) if p: return f"{cls.backend_description} • {get_serial_port_label(p)}" else: return f"{cls.backend_description} • {port}" @classmethod def get_switcher_entries(cls): def should_show(conf): port = conf[f"{cls.backend_name}.port"] if port == WEBREPL_PORT_VALUE: return True elif port == "auto": potential_ports = cls._detect_potential_ports() return len(potential_ports) > 0 else: for p in list_serial_ports(): if p.device == port: return True return False relevant_confs = list(filter(should_show, cls.get_last_configurations())) for device, desc in cls._detect_potential_ports(): conf = {"run.backend_name": cls.backend_name, f"{cls.backend_name}.port": device} if conf not in relevant_confs: relevant_confs.append(conf) sorted_confs = sorted(relevant_confs, key=cls.get_switcher_configuration_label) return [(conf, cls.get_switcher_configuration_label(conf)) for conf in sorted_confs] def has_custom_system_shell(self): return self._port and self._port != WEBREPL_PORT_VALUE def open_custom_system_shell(self): from thonny import terminal get_runner().send_command_and_wait(InlineCommand("prepare_disconnect"), "Disconnecting") self.disconnect() terminal.run_in_terminal( [ running.get_front_interpreter_for_subprocess(sys.executable), "-m", # "serial.tools.miniterm", "thonny.plugins.micropython.miniterm_wrapper", "--exit-char", "20", "--menu-char", "29", "--filter", "direct", "--quiet", self._port, "115200", ], cwd=get_workbench().get_local_cwd(), keep_open=False, title=self._port, ) def run_script_in_terminal(self, script_path, args, interactive, keep_open): messagebox.showinfo( "Running in terminal", "In order to run your script in terminal, save it on the device\n" "as main script, select 'Tools => Open system shell' and press Ctrl+D", ) class BareMetalMicroPythonConfigPage(BackendDetailsConfigPage): backend_name = None # Will be overwritten on Workbench.add_backend def __init__(self, master): super().__init__(master) self._has_opened_python_flasher = False intro_text = self._get_intro_text() if intro_text: intro_label = ttk.Label(self, text=intro_text) intro_label.grid(row=0, column=0, sticky="nw") intro_url = self._get_intro_url() if intro_url: intro_url_label = create_url_label(self, intro_url) intro_url_label.grid(row=1, column=0, sticky="nw") port_label = ttk.Label( self, text=tr("Port or WebREPL") if self.allow_webrepl else tr("Port") ) port_label.grid(row=3, column=0, sticky="nw", pady=(10, 0)) self._ports_by_desc = {get_serial_port_label(p): p.device for p in list_serial_ports()} self._ports_by_desc["< " + tr("Try to detect port automatically") + " >"] = "auto" self._WEBREPL_OPTION_DESC = "< WebREPL >" if self.allow_webrepl: self._ports_by_desc[self._WEBREPL_OPTION_DESC] = WEBREPL_PORT_VALUE def port_order(p): _, name = p if name is None: return "" elif name.startswith("COM") and len(name) == 4: # Make one-digit COM ports go before COM10 return name.replace("COM", "COM0") else: return name # order by port, auto first port_descriptions = [key for key, _ in sorted(self._ports_by_desc.items(), key=port_order)] self._port_desc_variable = create_string_var( self.get_stored_port_desc(), self._on_change_port ) self._port_combo = ttk.Combobox( self, exportselection=False, textvariable=self._port_desc_variable, values=port_descriptions, ) self._port_combo.state(["!disabled", "readonly"]) self._port_combo.grid(row=4, column=0, sticky="new") self.columnconfigure(0, weight=1) self._webrepl_frame = None self.add_checkbox( self.backend_name + ".interrupt_on_connect", row=10, description=tr("Interrupt working program on connect"), pady=(ems_to_pixels(2.0), 0), ) if self.may_have_rtc(): self.add_checkbox( self.backend_name + ".sync_time", row=11, description=tr("Synchronize device's real time clock"), ) self.add_checkbox( self.backend_name + ".local_rtc", row=12, description=tr("Use local time in real time clock"), ) self.add_checkbox( self.backend_name + ".restart_interpreter_before_run", row=13, description=tr("Restart interpreter before running a script"), ) last_row = ttk.Frame(self) last_row.grid(row=100, sticky="swe") self.rowconfigure(100, weight=1) last_row.columnconfigure(1, weight=1) kinds = self.get_flashing_dialog_kinds() for i, kind in enumerate(kinds): def _click_flashing_link(event, kind=kind): self._handle_python_installer_link(kind=kind) if i == 0: link_text = self._get_flasher_link_title() else: link_text = "" if kind != "": if link_text: link_text += " " link_text += f"({kind})" python_link = ui_utils.create_action_label( last_row, link_text, _click_flashing_link, ) python_link.grid(row=i, column=1, sticky="e") self._on_change_port() def _get_flasher_link_title(self) -> str: return tr("Install or update %s") % "MicroPython" def _handle_python_installer_link(self, kind: str): self._open_flashing_dialog(kind) self._has_opened_python_flasher = True def _get_intro_text(self): result = ( tr("Connect your device to the computer and select corresponding port below") + "\n" + "(" + tr('look for your device name, "USB Serial" or "UART"') + ").\n" + tr("If you can't find it, you may need to install proper USB driver first.") ) if self.allow_webrepl: result = ( ("Connecting via USB cable:") + "\n" + result + "\n\n" + ("Connecting via WebREPL:") + "\n" + ( "If your device supports WebREPL, first connect via serial, make sure WebREPL is enabled\n" + "(import webrepl_setup), connect your computer and device to same network and select\n" + "< WebREPL > below" ) ) return result def _get_webrepl_frame(self): if self._webrepl_frame is not None: return self._webrepl_frame self._webrepl_frame = ttk.Frame(self) self._webrepl_url_var = create_string_var( get_workbench().get_option(self.backend_name + ".webrepl_url") ) url_label = ttk.Label(self._webrepl_frame, text="URL (eg. %s)" % DEFAULT_WEBREPL_URL) url_label.grid(row=0, column=0, sticky="nw", pady=(10, 0)) url_entry = ttk.Entry(self._webrepl_frame, textvariable=self._webrepl_url_var, width=30) url_entry.grid(row=1, column=0, sticky="nw") self._webrepl_password_var = create_string_var( get_workbench().get_option(self.backend_name + ".webrepl_password") ) pw_label = ttk.Label(self._webrepl_frame, text=tr("Password")) pw_label.grid(row=0, column=1, sticky="nw", pady=(10, 0), padx=(10, 0)) pw_entry = ttk.Entry(self._webrepl_frame, textvariable=self._webrepl_password_var, width=15) pw_entry.grid(row=1, column=1, sticky="nw", padx=(10, 0)) return self._webrepl_frame def get_stored_port_desc(self): name = get_workbench().get_option(self.backend_name + ".port") for desc in self._ports_by_desc: if self._ports_by_desc[desc] == name: return desc return "" def get_selected_port_name(self): port_desc = self._port_desc_variable.get() if not port_desc: return None return self._ports_by_desc[port_desc] def is_modified(self): return ( self._port_desc_variable.modified or self.webrepl_selected() and self._webrepl_password_var.modified or self.webrepl_selected() and self._webrepl_url_var.modified ) def webrepl_selected(self): return self.get_selected_port_name() == WEBREPL_PORT_VALUE def should_restart(self): return self.is_modified() or self._has_opened_python_flasher def apply(self): if not self.is_modified(): return else: port_name = self.get_selected_port_name() get_workbench().set_option(self.backend_name + ".port", port_name) if self.webrepl_selected(): if not self._webrepl_url_var.get().lower().startswith("ws://"): messagebox.showerror( "Bad URL", "WebREPL URL should start with ws://", parent=self ) return False get_workbench().set_option( self.backend_name + ".webrepl_url", self._webrepl_url_var.get() ) get_workbench().set_option( self.backend_name + ".webrepl_password", self._webrepl_password_var.get() ) def _on_change_port(self, *args): if self._port_desc_variable.get() == self._WEBREPL_OPTION_DESC: self._get_webrepl_frame().grid(row=6, column=0, sticky="nwe") else: if self._webrepl_frame and self._webrepl_frame.winfo_ismapped(): self._webrepl_frame.grid_forget() def may_have_rtc(self): return True def _get_intro_url(self) -> Optional[str]: return None def get_flashing_dialog_kinds(self) -> List[str]: return [] def _open_flashing_dialog(self, kind: str) -> None: raise NotImplementedError() @property def allow_webrepl(self): return False class GenericBareMetalMicroPythonProxy(BareMetalMicroPythonProxy): @classmethod def get_known_usb_vids_pids(cls): """Return set of pairs of USB device (VID, PID)""" return { # Generic MicroPython Board, see http://pid.codes/org/MicroPython/ (0x1209, 0xADDA), (0x0694, 0x0009), # SPIKE Prime (0x0694, 0x0010), # Robot Inventor } | get_uart_adapter_vids_pids() @classmethod def get_vids_pids_to_avoid(self): return VIDS_PIDS_TO_AVOID_IN_GENERIC_BACKEND class GenericBareMetalMicroPythonConfigPage(BareMetalMicroPythonConfigPage): @property def allow_webrepl(self): return False class LocalMicroPythonProxy(MicroPythonProxy): def __init__(self, clean): self._target_executable = get_workbench().get_option("LocalMicroPython.executable") super().__init__(clean) def get_target_executable(self) -> Optional[str]: return self._target_executable def _get_launcher_with_args(self): import thonny.plugins.micropython.os_mp_backend cmd = [ thonny.plugins.micropython.os_mp_backend.__file__, repr( { "interpreter": self._target_executable, "cwd": self.get_cwd(), } ), ] return cmd def interrupt(self): # Don't interrupt local process, but direct it to the device self._send_msg(ImmediateCommand("interrupt")) def send_command(self, cmd: CommandToBackend) -> Optional[str]: if isinstance(cmd, EOFCommand): # Runner doesn't notice restart get_shell().restart(was_running=get_runner().is_running()) return super().send_command(cmd) def _get_initial_cwd(self): return get_workbench().get_local_cwd() def supports_remote_files(self): return False # return self._proc is not None def uses_local_filesystem(self): return True def ready_for_remote_file_operations(self): return False def supports_remote_directories(self): return False def supports_trash(self): return True def is_connected(self): return self._proc is not None def _show_error(self, text): get_shell().print_error("\n" + text + "\n") def disconnect(self): self.destroy() def get_node_label(self): return "Local" def get_full_label(self): return self._target_executable def get_exe_dirs(self): return [] def can_run_local_files(self): return True def can_run_remote_files(self): return False @classmethod def _get_switcher_conf_for_executable(cls, executable): return {"run.backend_name": cls.backend_name, f"{cls.backend_name}.executable": executable} def get_current_switcher_configuration(self): return self._get_switcher_conf_for_executable( get_workbench().get_option(f"{self.backend_name}.executable") ) @classmethod def get_switcher_entries(cls): confs = sorted( cls.get_last_configurations(), key=lambda conf: conf[f"{cls.backend_name}.executable"] ) return [ (conf, cls.get_switcher_configuration_label(conf)) for conf in confs if os.path.exists(conf[f"{cls.backend_name}.executable"]) or shutil.which(conf[f"{cls.backend_name}.executable"]) ] @classmethod def get_switcher_configuration_label(cls, conf: Dict[str, Any]) -> str: return cls.backend_description + " • " + conf[f"{cls.backend_name}.executable"] @classmethod def is_valid_configuration(cls, conf: Dict[str, Any]) -> bool: executable = conf[f"{cls.backend_name}.executable"] return os.path.exists(executable) or shutil.which(executable) class LocalMicroPythonConfigPage(BackendDetailsConfigPage): backend_name = None # Will be overwritten on Workbench.add_backend def __init__(self, master): super().__init__(master) self._changed = False self._executable_var = self._add_text_field( "Interpreter", "LocalMicroPython.executable", 30 ) def _on_change(self): self._changed = True def apply(self): get_workbench().set_option("LocalMicroPython.executable", self._executable_var.get()) def should_restart(self): return self._changed class SshMicroPythonProxy(MicroPythonProxy): def __init__(self, clean): self._host = get_workbench().get_option(f"{self.backend_name}.host") self._user = get_workbench().get_option(f"{self.backend_name}.user") self._target_executable = get_workbench().get_option(f"{self.backend_name}.executable") super().__init__(clean) def _get_launcher_with_args(self): import thonny.plugins.micropython.os_mp_backend args = { "cwd": get_workbench().get_option(f"{self.backend_name}.cwd") or "", "interpreter": self._target_executable, "host": self._host, "user": self._user, } args.update(self._get_time_args()) args.update(self._get_extra_launcher_args()) cmd = [ thonny.plugins.micropython.os_mp_backend.__file__, repr(args), ] return cmd def _send_initial_input(self) -> None: assert self._proc is not None self._proc.stdin.write((get_ssh_password(self.backend_name) or "") + "\n") self._proc.stdin.flush() def _get_extra_launcher_args(self): return {} def get_target_executable(self) -> Optional[str]: return self._target_executable def interrupt(self): # Don't interrupt local process, but direct it to the device self._send_msg(ImmediateCommand("interrupt")) def send_command(self, cmd: CommandToBackend) -> Optional[str]: if isinstance(cmd, EOFCommand): # Runner doesn't notice restart get_shell().restart(was_running=get_runner().is_running()) return super().send_command(cmd) def _get_initial_cwd(self): return get_workbench().get_option(f"{self.backend_name}.cwd") def _publish_cwd(self, cwd): return get_workbench().set_option(f"{self.backend_name}.cwd", cwd) def supports_remote_files(self): return True # return self._proc is not None def uses_local_filesystem(self): return False def ready_for_remote_file_operations(self): return self.is_connected() def supports_remote_directories(self): return True def supports_trash(self): return False def is_connected(self): return self._proc is not None def _show_error(self, text): get_shell().print_error("\n" + text + "\n") def disconnect(self): self.destroy() def get_node_label(self): return self._host def get_full_label(self): return self._target_executable + " @ " + self._host def get_exe_dirs(self): return [] def can_run_local_files(self): return False def can_run_remote_files(self): return True def get_current_switcher_configuration(self) -> Dict[str, Any]: return { "run.backend_name": self.backend_name, f"{self.backend_name}.executable": get_workbench().get_option( f"{self.backend_name}.executable" ), f"{self.backend_name}.host": get_workbench().get_option(f"{self.backend_name}.host"), f"{self.backend_name}.user": get_workbench().get_option(f"{self.backend_name}.user"), } @classmethod def get_switcher_configuration_label(cls, conf: Dict[str, Any]) -> str: user = conf[f"{cls.backend_name}.user"] host = conf[f"{cls.backend_name}.host"] executable = conf[f"{cls.backend_name}.executable"] return f"{cls.backend_description} • {user} @ {host} : {executable}" @classmethod def get_switcher_entries(cls): confs = sorted(cls.get_last_configurations(), key=cls.get_switcher_configuration_label) return [(conf, cls.get_switcher_configuration_label(conf)) for conf in confs] def has_custom_system_shell(self): return True def open_custom_system_shell(self): if not shutil.which("ssh"): messagebox.showerror( "Command not found", "Command 'ssh' not found", master=get_workbench() ) return from thonny import terminal userhost = "%s@%s" % (self._user, self._host) terminal.run_in_terminal( ["ssh", userhost], cwd=get_workbench().get_local_cwd(), keep_open=False, title=userhost ) @classmethod def is_valid_configuration(cls, conf: Dict[str, Any]) -> bool: return True class SshMicroPythonConfigPage(BaseSshProxyConfigPage): pass _PORTS_CACHE = [] _PORTS_CACHE_TIME = 0 def get_serial_port_label(p) -> str: # On Windows, port is given also in description if p.product: desc = p.product elif p.interface: desc = p.interface else: desc = p.description.replace(f" ({p.device})", "") if desc == "USB Serial Device": # Try finding something less generic if p.product: desc = p.product elif p.interface: desc = p.interface return f"{desc} @ {p.device}" def list_serial_ports(max_cache_age: float = 0.5, skip_logging: bool = False): global _PORTS_CACHE, _PORTS_CACHE_TIME cur_time = time.time() if cur_time - _PORTS_CACHE_TIME > max_cache_age: _PORTS_CACHE = _list_serial_ports_uncached(skip_logging=skip_logging) _PORTS_CACHE_TIME = cur_time return _PORTS_CACHE def _list_serial_ports_uncached(skip_logging: bool = False): if not skip_logging: logger.info("Listing serial ports") # serial.tools.list_ports.comports() can be too slow # because os.path.islink can be too slow (https://github.com/pyserial/pyserial/pull/303) # Workarond: temporally patch os.path.islink old_islink = os.path.islink try: if sys.platform == "win32": os.path.islink = lambda _: False if sys.platform == "win32": try: from adafruit_board_toolkit._list_ports_windows import comports except ImportError: logger.info("Falling back to serial.tools.list_ports.comports") from serial.tools.list_ports import comports elif sys.platform == "darwin": try: from adafruit_board_toolkit._list_ports_osx import comports except ImportError: logger.info("Falling back to serial.tools.list_ports.comports") from serial.tools.list_ports import comports else: from serial.tools.list_ports import comports irrelevant = ["/dev/cu.Bluetooth-Incoming-Port", "/dev/cu.iPhone-WirelessiAP"] result = [] for p in comports(): if p.device not in irrelevant: result.append(p) return result finally: os.path.islink = old_islink def port_exists(device): for port in list_serial_ports(): if port.device == device: return True return False def get_uart_adapter_vids_pids(): # https://github.com/per1234/zzInoVIDPID # https://github.com/per1234/zzInoVIDPID/blob/master/zzInoVIDPID/boards.txt # http://esp32.net/usb-uart/ # https://www.usb.org/developers # https://github.com/espressif/usb-pids return { (0x1A86, 0x7523), # CH340 (HL-340?) (0x1A86, 0x5523), # CH341 (0x1A86, 0x55D4), # CH9102F, seen at Adafruit Feather ESP32 V2, M5 stamp C3 (0x10C4, 0xEA60), # CP210x, (0x0403, 0x6001), # FT232/FT245 (XinaBox CW01, CW02) (0x0403, 0x6010), # FT2232C/D/L/HL/Q (ESP-WROVER-KIT) (0x0403, 0x6011), # FT4232 (0x0403, 0x6014), # FT232H (0x0403, 0x6015), # FT X-Series (Sparkfun ESP32) (0x0403, 0x601C), # FT4222H (0x303A, 0x1001), # Espressif's built-in USB-to-Serial, seen at QtPy C3 } def get_port_info(port): for info in list_serial_ports(): if info.device == port: return info raise RuntimeError("Port %s not found" % port) def add_micropython_backend( name, proxy_class, description, config_page, bare_metal=True, sort_key=None, validate_time=False, sync_time=None, local_rtc=True, submit_mode=None, write_block_size=None, write_block_delay=None, dtr=None, rts=None, ): if bare_metal: get_workbench().set_default(name + ".port", "auto") get_workbench().set_default(name + ".webrepl_url", DEFAULT_WEBREPL_URL) get_workbench().set_default(name + ".webrepl_password", "") get_workbench().set_default(name + ".submit_mode", submit_mode) get_workbench().set_default(name + ".write_block_size", write_block_size) get_workbench().set_default(name + ".write_block_delay", write_block_delay) get_workbench().set_default(name + ".dtr", dtr) get_workbench().set_default(name + ".rts", rts) get_workbench().set_default(name + ".interrupt_on_connect", True) get_workbench().set_default(name + ".restart_interpreter_before_run", True) if sync_time is None: sync_time = True else: if sync_time is None: sync_time = False get_workbench().set_default(name + ".sync_time", sync_time) get_workbench().set_default(name + ".local_rtc", local_rtc) get_workbench().set_default(name + ".validate_time", validate_time) get_workbench().add_backend(name, proxy_class, description, config_page, sort_key=sort_key)