feat: add pyOCD target override support and enhance firmware download logic

Signed-off-by: Jos Verlinde <Jos_Verlinde@hotmail.com>

Author: Josverl

mpflash/cli_flash.py

@@ -135,6 +135,13 @@
     default=True,
     help="""Automatically install CMSIS packs for missing pyOCD targets. Default: enabled.""",
 )
+@click.option(
+    "--target",
+    "pyocd_target",
+    default=None,
+    help="""Explicit pyOCD target override (for --method pyocd), e.g. rp2040 or r7fa4m1ab.""",
+    metavar="PYOCD_TARGET",
+)
 @click.option(
     "--force",
     "-f",
@@ -225,6 +232,7 @@ def _create_worklist_or_fail(*create_args, **create_kwargs) -> FlashTaskList:
     # Extract pyOCD options
     probe_id = kwargs.pop("probe_id", None)
     auto_install_packs = kwargs.pop("auto_install_packs", True)
+    pyocd_target = kwargs.pop("pyocd_target", None)
 
     params = FlashParams(**kwargs)
     params.versions = list(params.versions)
@@ -383,6 +391,7 @@ def _create_worklist_or_fail(*create_args, **create_kwargs) -> FlashTaskList:
         method=flash_method,
         probe_id=probe_id,
         auto_install_packs=auto_install_packs,
+        target_override=pyocd_target,
         flash_mode=params.flash_mode,
         retry_on_error=params.retry_on_error,
         retry_baud=params.retry_baud,

mpflash/flash/__init__.py

@@ -54,6 +54,8 @@ def flash_tasks(
 ):
     """Flash every entry in ``tasks`` and return the updated boards."""
 
+    attempted_backend_downloads: set[tuple[str, str, str, str, bool]] = set()
+
     def _pick_backend_compatible_firmware(task, fw_info):
         """Pick a firmware image matching the explicit backend's supported formats."""
         if fw_info is None:
@@ -89,34 +91,88 @@ def _pick_backend_compatible_firmware(task, fw_info):
         detected_board_id = f"{board.board}-{board.variant}" if board.variant else board.board
         board_ids = [getattr(fw_info, "board_id", ""), detected_board_id]
 
-        candidates = []
-        seen_files = set()
-        for bid in board_ids:
-            if not bid:
-                continue
-            # First prefer exact port match, then broaden to any port.
-            for cand in find_downloaded_firmware(
-                board_id=bid,
-                version=fw_info.version,
-                port=board.port,
-                custom=bool(fw_info.custom),
-            ) + find_downloaded_firmware(
-                board_id=bid,
-                version=fw_info.version,
-                port="",
-                custom=bool(fw_info.custom),
-            ):
-                if cand.firmware_file not in seen_files:
-                    seen_files.add(cand.firmware_file)
-                    candidates.append(cand)
-
-        for cand in reversed(candidates):
-            if Path(cand.firmware_file).suffix.lower() in backend.supported_formats:
+        def _find_supported_candidate():
+            candidates = []
+            seen_files = set()
+            for bid in board_ids:
+                if not bid:
+                    continue
+                # First prefer exact port match, then broaden to any port.
+                for cand in find_downloaded_firmware(
+                    board_id=bid,
+                    version=fw_info.version,
+                    port=board.port,
+                    custom=bool(fw_info.custom),
+                ) + find_downloaded_firmware(
+                    board_id=bid,
+                    version=fw_info.version,
+                    port="",
+                    custom=bool(fw_info.custom),
+                ):
+                    if cand.firmware_file not in seen_files:
+                        seen_files.add(cand.firmware_file)
+                        candidates.append(cand)
+
+            for cand in reversed(candidates):
+                if Path(cand.firmware_file).suffix.lower() in backend.supported_formats:
+                    return cand
+            return None
+
+        if candidate := _find_supported_candidate():
+            log.info(
+                f"Using {requested_name} compatible firmware {candidate.firmware_file} "
+                f"instead of {fw_info.firmware_file} for {board.board} on {board.serialport}"
+            )
+            return candidate
+
+        # No local firmware matches backend-supported file types; try one
+        # targeted download refresh before handing off to backend selection.
+        download_key = (
+            requested_name,
+            detected_board_id,
+            fw_info.version,
+            board.port or "",
+            bool(fw_info.custom),
+        )
+        if download_key not in attempted_backend_downloads:
+            attempted_backend_downloads.add(download_key)
+            try:
+                from mpflash.download import download
+                from mpflash.mpboard_id.alternate import alternate_board_names
+
+                log.info(
+                    f"No local {requested_name} firmware with suffix in "
+                    f"{list(backend.supported_formats)} for {board.board} on {board.serialport}; "
+                    "trying firmware download refresh"
+                )
+                download(
+                    ports=[board.port] if board.port else [],
+                    boards=alternate_board_names(detected_board_id, board.port),
+                    versions=[fw_info.version],
+                    force=True,
+                    clean=True,
+                )
+            except Exception as exc:  # noqa: BLE001 - fallback to original firmware below
+                log.debug(
+                    f"Backend-compatible firmware refresh failed for {detected_board_id} "
+                    f"{fw_info.version}: {exc}"
+                )
+
+            if candidate := _find_supported_candidate():
                 log.info(
-                    f"Using {requested_name} compatible firmware {cand.firmware_file} "
+                    f"Using downloaded {requested_name} compatible firmware {candidate.firmware_file} "
                     f"instead of {fw_info.firmware_file} for {board.board} on {board.serialport}"
                 )
-                return cand
+                return candidate
+
+        raise MPFlashError(
+            f"No firmware matching backend {requested_name!r} for "
+            f"{detected_board_id!r} {fw_info.version}. "
+            f"Selected firmware is {fw_info.firmware_file!r} ({current_suffix or '<none>'}), "
+            f"but {requested_name} supports {list(backend.supported_formats)}. "
+            "A download refresh was attempted but no compatible firmware was found."
+        )
+
         return fw_info
 
     flashed = []
@@ -174,12 +230,38 @@ def flash_mcu(
     :attr:`FlashContext.options` for the backend to consume.
     """
     requested = _resolve_backend_name(method)
-    try:
-        backend = select_backend(mcu, fw_file, requested_name=requested)
-    except MPFlashError:
-        raise
-    except Exception as e:  # noqa: BLE001 - selection should never crash callers
-        raise MPFlashError(f"Failed to select flash backend: {e}") from e
+    target_override = kwargs.get("target_override")
+
+    # When the user explicitly provides a pyOCD target override, skip
+    # capability probing that depends on MCU metadata-derived target matching.
+    if requested == "pyocd" and target_override:
+        backend = get_backend("pyocd")
+        if backend is None:
+            raise MPFlashError("Unknown flash method 'pyocd'.")
+        platform = default_services.current_platform()
+        suffix = fw_file.suffix.lower()
+        if backend.supported_formats and suffix not in backend.supported_formats:
+            raise MPFlashError(
+                f"Backend 'pyocd' cannot flash {fw_file.name}: unsupported format {suffix or '<none>'}."
+            )
+        if backend.supported_platforms and platform not in backend.supported_platforms:
+            raise MPFlashError(
+                f"Backend 'pyocd' does not run on {platform.value}."
+            )
+        if not backend.is_available():
+            raise MPFlashError(
+                "pyOCD is not installed (install with: uv sync --extra pyocd)."
+            )
+        log.info(
+            f"Using explicit pyOCD target override '{target_override}' for {mcu.board_id or mcu.board}"
+        )
+    else:
+        try:
+            backend = select_backend(mcu, fw_file, requested_name=requested)
+        except MPFlashError:
+            raise
+        except Exception as e:  # noqa: BLE001 - selection should never crash callers
+            raise MPFlashError(f"Failed to select flash backend: {e}") from e
 
     log.debug(f"Using flash backend: {backend.name} for {mcu.board_id}")
 

mpflash/flash/builtins/pyocd/core.py

@@ -355,15 +355,16 @@ def get_pyocd_targets() -> Dict[str, Dict[str, str]]:
 
                     subprocess_targets[target_name] = {"vendor": vendor, "part_number": part_number, "source": source}
 
-            # Merge subprocess results (subprocess is authoritative)
-            if len(subprocess_targets) > len(targets):
-                targets = subprocess_targets
-                log.debug(f"Subprocess method loaded {len(targets)} total targets")
-            else:
-                # Supplement API results with any pack targets from subprocess
-                pack_targets = {k: v for k, v in subprocess_targets.items() if v["source"] == "pack" and k not in targets}
-                targets.update(pack_targets)
-                log.debug(f"Added {len(pack_targets)} pack targets from subprocess")
+            # Merge API + subprocess results.
+            # Some pyOCD installations omit certain builtins in CLI output
+            # (RP2040 is one observed example), so never discard API targets.
+            if subprocess_targets:
+                added = len([k for k in subprocess_targets if k not in targets])
+                targets.update(subprocess_targets)
+                log.debug(
+                    f"Merged {len(subprocess_targets)} subprocess targets "
+                    f"({added} new, {len(subprocess_targets) - added} updated)"
+                )
 
     except Exception as subprocess_error:
         log.debug(f"Subprocess target discovery failed: {subprocess_error}")
@@ -631,6 +632,32 @@ def detect_pyocd_target(mcu: MPRemoteBoard, auto_install_packs: bool = True) ->
             _target_cache[cache_key] = target
             return target
 
+        # RP2 fallback: pyOCD supports RP2040/RP2350 families but target names
+        # may vary across versions (rp2040, rp2040_core0, rp2350, ...).
+        if not target and mcu_info.get("port", "").lower() == "rp2":
+            board_id = (mcu.board_id or "").upper()
+            cpu = (mcu.cpu or "").upper()
+            if "RP2350" in cpu or "PICO2" in board_id:
+                rp_family = "rp2350"
+            else:
+                rp_family = "rp2040"
+
+            # Prefer exact family target, then core0, then any same-family target.
+            if rp_family in pyocd_targets:
+                target = rp_family
+            elif f"{rp_family}_core0" in pyocd_targets:
+                target = f"{rp_family}_core0"
+            else:
+                for candidate in pyocd_targets:
+                    if candidate.lower().startswith(rp_family):
+                        target = candidate
+                        break
+
+            if target:
+                log.info(f"RP2 fallback target match: {mcu.board_id} -> {target}")
+                _target_cache[cache_key] = target
+                return target
+
         # No target found - try automatic pack installation if enabled
         if auto_install_packs and chip_family:
             log.info(f"No pyOCD target found for {chip_family}, attempting automatic pack installation")

mpflash/flash/builtins/pyocd/flash.py

@@ -435,20 +435,33 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 class PyOCDFlash:
     """High-level pyOCD flash programming interface."""
 
-    def __init__(self, mcu: MPRemoteBoard, probe_id: Optional[str] = None, auto_install_packs: bool = True):
+    def __init__(
+        self,
+        mcu: MPRemoteBoard,
+        probe_id: Optional[str] = None,
+        auto_install_packs: bool = True,
+        target_override: Optional[str] = None,
+    ):
         """
         Initialize PyOCD flash programmer.
 
         Args:
             mcu: MPRemoteBoard instance with board information
             probe_id: Specific probe unique ID to use (optional)
             auto_install_packs: Automatically install missing CMSIS packs
+            target_override: Explicit pyOCD target name override (optional)
         """
         self.mcu = mcu
         self.probe_id = probe_id
 
-        # Detect target type using core functionality
-        self.target_type = detect_pyocd_target(mcu, auto_install_packs=auto_install_packs)
+        # Detect target type using core functionality unless explicitly overridden.
+        self.target_type = target_override or detect_pyocd_target(
+            mcu, auto_install_packs=auto_install_packs
+        )
+        if target_override:
+            log.info(
+                f"Using explicit pyOCD target override for {mcu.board_id}: {target_override}"
+            )
 
         if not is_pyocd_available():
             raise MPFlashError("No debug probe support available. Install with: uv sync --extra pyocd")
@@ -614,7 +627,13 @@ def find_pyocd_probe(probe_id: Optional[str] = None) -> Optional[PyOCDProbe]:
 
 
 def flash_pyocd(
-    mcu: MPRemoteBoard, fw_file: Path, erase: bool = False, probe_id: Optional[str] = None, auto_install_packs: bool = True, **kwargs
+    mcu: MPRemoteBoard,
+    fw_file: Path,
+    erase: bool = False,
+    probe_id: Optional[str] = None,
+    auto_install_packs: bool = True,
+    target_override: Optional[str] = None,
+    **kwargs,
 ) -> bool:
     """
     Flash MCU using pyOCD SWD/JTAG interface.
@@ -625,6 +644,7 @@ def flash_pyocd(
         erase: Whether to erase flash before programming
         probe_id: Specific debug probe ID to use (optional)
         auto_install_packs: Automatically install missing CMSIS packs
+        target_override: Explicit pyOCD target name override (optional)
         **kwargs: Additional options
 
     Returns:
@@ -633,12 +653,17 @@ def flash_pyocd(
     Raises:
         MPFlashError: If flashing fails
     """
-    if not is_pyocd_supported(mcu):
+    if not target_override and not is_pyocd_supported(mcu):
         reason = get_unsupported_reason(mcu)
         raise MPFlashError(f"PyOCD flash not supported: {reason}")
 
     # Create flasher and program
-    flasher = PyOCDFlash(mcu, probe_id=probe_id, auto_install_packs=auto_install_packs)
+    flasher = PyOCDFlash(
+        mcu,
+        probe_id=probe_id,
+        auto_install_packs=auto_install_packs,
+        target_override=target_override,
+    )
     ok = flasher.flash_firmware(fw_file, erase=erase, **kwargs)
     if ok:
         # Give the board a moment to reset and re-enumerate over USB,

mpflash/flash/builtins/pyocd_backend.py

@@ -76,7 +76,7 @@ def flash(self, ctx: FlashContext) -> FlashResult:
 
         passthrough = {
             k: ctx.options[k]
-            for k in ("probe_id", "auto_install_packs")
+            for k in ("probe_id", "auto_install_packs", "target_override")
             if k in ctx.options
         }
         ok = flash_pyocd(ctx.mcu, fw_file=ctx.fw_file, erase=ctx.erase, **passthrough)