Merge pull request #554 from Josverl/dev/mpflash_board_id

.github/workflows/pytest.yml

@@ -42,6 +42,7 @@ jobs:
             matrix:
                 os: [ubuntu-latest, windows-latest, macos-13]
                 python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+                mpflash-version: ["beta"]
                 include: # for testing
                     - os: windows-latest
                       python-version: "3.11"
@@ -88,8 +89,13 @@ jobs:
               run: |
                   poetry install --with test
                   # use the latest source version of mpflash
-                  pip uninstall mpflash
-                  pip install src/mpflash
+
+            - name: update mpflash to beta for testing
+              if: ${{ matrix.mpflash-version == 'beta' }}
+              run: |
+                  # use poetry's environmnt
+                  poetry run pip uninstall mpflash -y
+                  poetry run pip install src/mpflash
 
             #----------------------------------------------
             # stubber clone

pyproject.toml

@@ -97,6 +97,7 @@ pytest-mock = "^3.10.0"
 mock = "^4.0.3"
 distro = "^1.8.0"
 fasteners = "^0.19"
+jsons = "^1.6.3"
 
 [tool.poetry.group.dev]
 optional = true

src/mpflash/mpflash/ask_input.py

@@ -12,7 +12,7 @@
 from loguru import logger as log
 
 from mpflash.config import config
-from mpflash.mpboard_id import get_stored_boards_for_port, known_stored_boards, local_mp_ports
+from mpflash.mpboard_id import get_known_boards_for_port, get_known_ports, known_stored_boards
 from mpflash.mpremoteboard import MPRemoteBoard
 from mpflash.vendor.versions import micropython_versions
 
@@ -79,22 +79,31 @@ def ask_missing_params(
 
     if not params.boards or "?" in params.boards:
         ask_port_board(questions, action=action)
-
-    answers = inquirer.prompt(questions, answers=answers)
+    if questions:
+        answers = inquirer.prompt(questions, answers=answers)
     if not answers:
         # input cancelled by user
         return []  # type: ignore
     # print(repr(answers))
     if isinstance(params, FlashParams) and "serial" in answers:
         params.serial = answers["serial"]
     if "port" in answers:
-        params.ports = [answers["port"]]
+        params.ports = [p for p in params.ports if p != "?"]  # remove the "?" if present
+        params.ports.extend(answers["port"])
     if "boards" in answers:
-        params.boards = answers["boards"] if isinstance(answers["boards"], list) else [answers["boards"]]
+        params.boards = [b for b in params.boards if b != "?"]  # remove the "?" if present
+        params.boards.extend(answers["boards"] if isinstance(answers["boards"], list) else [answers["boards"]])
     if "versions" in answers:
+        params.versions = [v for v in params.versions if v != "?"]  # remove the "?" if present
         # make sure it is a list
-        params.versions = answers["versions"] if isinstance(answers["versions"], list) else [answers["versions"]]
-
+        if isinstance(answers["versions"], (list, tuple)):
+            params.versions.extend(answers["versions"])
+        else:
+            params.versions.append(answers["versions"])
+    # remove duplicates
+    params.ports = list(set(params.ports))
+    params.boards = list(set(params.boards))
+    params.versions = list(set(params.versions))
     log.debug(repr(params))
 
     return params
@@ -149,14 +158,15 @@ def ask_port_board(questions: list, *, action: str):
     # import only when needed to reduce load time
     import inquirer
 
-    # TODO: if action = flash, Use Inquirer.List for boards
+    # if action flash,  single input
+    # if action download, multiple input
     inquirer_ux = inquirer.Checkbox if action == "download" else inquirer.List
     questions.extend(
         (
             inquirer.List(
                 "port",
                 message="Which port do you want to {action} " + "to {serial} ?" if action == "flash" else "?",
-                choices=local_mp_ports(),
+                choices=get_known_ports(),
                 autocomplete=True,
             ),
             inquirer_ux(
@@ -194,7 +204,7 @@ def ask_versions(questions: list, *, action: str):
 
     # remove the versions for which there are no known boards in the board_info.json
     # todo: this may be a little slow
-    mp_versions = [v for v in mp_versions if get_stored_boards_for_port("stm32", [v])]
+    mp_versions = [v for v in mp_versions if get_known_boards_for_port("stm32", [v])]
 
     mp_versions.append("preview")
     mp_versions.reverse()  # newest first

src/mpflash/mpflash/cli_download.py

@@ -7,7 +7,7 @@
 from loguru import logger as log
 
 from mpflash.errors import MPFlashError
-from mpflash.mpboard_id import find_stored_board
+from mpflash.mpboard_id import find_known_board
 from mpflash.vendor.versions import clean_version
 
 from .ask_input import DownloadParams, ask_missing_params
@@ -68,10 +68,17 @@ def cli_download(**kwargs) -> int:
     params.versions = list(params.versions)
     params.boards = list(params.boards)
     if params.boards:
-        pass
-        # TODO Clean board - same as in cli_flash.py
+        if not params.ports:
+            # no ports specified - resolve ports from specified boards by resolving board IDs
+            for board in params.boards:
+                if board != "?":
+                    try:
+                        board_ = find_known_board(board)
+                        params.ports.append(board_["port"])
+                    except MPFlashError as e:
+                        log.error(f"{e}")
     else:
-        # no boards specified - detect connected boards
+        # no boards specified - detect connected ports and boards
         params.ports, params.boards = connected_ports_boards()
 
     params = ask_missing_params(params, action="download")

src/mpflash/mpflash/cli_flash.py

@@ -4,7 +4,7 @@
 from loguru import logger as log
 
 from mpflash.errors import MPFlashError
-from mpflash.mpboard_id import find_stored_board
+from mpflash.mpboard_id import find_known_board
 from mpflash.vendor.versions import clean_version
 
 from .ask_input import FlashParams, ask_missing_params
@@ -116,7 +116,7 @@ def cli_flash_board(**kwargs) -> int:
                 continue
             if " " in board_id:
                 try:
-                    info = find_stored_board(board_id)
+                    info = find_known_board(board_id)
                     if info:
                         log.info(f"Resolved board description: {info['board']}")
                         params.boards.remove(board_id)

src/mpflash/mpflash/download.py

@@ -20,6 +20,7 @@
 
 from mpflash.common import PORT_FWTYPES
 from mpflash.errors import MPFlashError
+from mpflash.mpboard_id import get_known_ports
 
 jsonlines.ujson = None  # type: ignore
 # #########################################################################################################
@@ -109,6 +110,8 @@ def get_boards(ports: List[str], boards: List[str], clean: bool) -> List[Firmwar
 
     """
     board_urls: List[FirmwareInfo] = []
+    if ports is None:
+        ports = get_known_ports()
     for port in ports:
         download_page_url = f"{MICROPYTHON_ORG_URL}download/?port={port}"
         _urls = get_board_urls(download_page_url)
@@ -170,6 +173,7 @@ def download_firmwares(
     skipped = downloaded = 0
     if versions is None:
         versions = []
+
     unique_boards = get_firmware_list(ports, boards, versions, clean)
 
     for b in unique_boards:

src/mpflash/mpflash/flash_esp.py

@@ -10,7 +10,7 @@
 import esptool
 from loguru import logger as log
 
-from mpflash.mpboard_id import find_stored_board
+from mpflash.mpboard_id import find_known_board
 from mpflash.mpremoteboard import MPRemoteBoard
 
 
@@ -22,7 +22,7 @@ def flash_esp(mcu: MPRemoteBoard, fw_file: Path, *, erase: bool = True) -> Optio
     log.info(f"Flashing {fw_file} on {mcu.board} on {mcu.serialport}")
     if not mcu.cpu:
         # Lookup CPU based on the board name
-        mcu.cpu = find_stored_board(mcu.board)["cpu"]
+        mcu.cpu = find_known_board(mcu.board)["cpu"]
 
     cmds: List[List[str]] = []
     if erase:

src/mpflash/mpflash/mpboard_id/__init__.py

@@ -9,10 +9,13 @@
 from pathlib import Path
 from typing import List, Optional, Tuple, TypedDict, Union
 
-from mpflash.errors import MPFlashError
 from mpflash.common import PORT_FWTYPES
+from mpflash.errors import MPFlashError
 from mpflash.vendor.versions import clean_version
 
+# KNOWN ports and boards are sourced from the micropython repo,
+# this info is stored in the board_info.json file
+
 
 # Board  based on the dataclass Board but changed to TypedDict
 # - source : get_boardnames.py
@@ -30,27 +33,27 @@ class Board(TypedDict):
 
 
 @lru_cache(maxsize=None)
-def read_stored_boardinfo() -> List[Board]:
+def read_known_boardinfo() -> List[Board]:
     """Reads the board_info.json file and returns the data as a list of Board objects"""
     with open(Path(__file__).parent / "board_info.json", "r") as file:
         return json.load(file)
 
 
-def local_mp_ports() -> List[str]:
+def get_known_ports() -> List[str]:
     # TODO: Filter for Version
-    mp_boards = read_stored_boardinfo()
+    mp_boards = read_known_boardinfo()
     # select the unique ports from info
     ports = set({board["port"] for board in mp_boards if board["port"] in PORT_FWTYPES.keys()})
     return sorted(list(ports))
 
 
-def get_stored_boards_for_port(port: str, versions: Optional[List[str]] = None):
+def get_known_boards_for_port(port: str, versions: Optional[List[str]] = None):
     """
     Returns a list of boards for the given port and version(s)
 
     port : str : The Micropython port to filter for
     versions : List[str] : The Micropython versions to filter for (actual versions required)"""
-    mp_boards = read_stored_boardinfo()
+    mp_boards = read_known_boardinfo()
 
     # filter for 'preview' as they are not in the board_info.json
     # instead use stable version
@@ -75,16 +78,16 @@ def known_stored_boards(port: str, versions: Optional[List[str]] = None) -> List
     port : str : The Micropython port to filter for
     versions : List[str] : The Micropython versions to filter for (actual versions required)
     """
-    mp_boards = get_stored_boards_for_port(port, versions)
+    mp_boards = get_known_boards_for_port(port, versions)
 
     boards = set({(f'{board["version"]} {board["description"]}', board["board"]) for board in mp_boards})
     return sorted(list(boards))
 
 
 @lru_cache(maxsize=20)
-def find_stored_board(board_id: str) -> Board:
-    """Find the board for the given board_ID or 'board description' and return the board info as a Board object"""
-    info = read_stored_boardinfo()
+def find_known_board(board_id: str) -> Board:
+    """Find the board for the given BOARD_ID or 'board description' and return the board info as a Board object"""
+    info = read_known_boardinfo()
     for board_info in info:
         if board_id in (board_info["board"], board_info["description"]):
             if "cpu" not in board_info or not board_info["cpu"]:

src/mpflash/mpflash/mpboard_id/board_id.py

@@ -8,7 +8,7 @@
 from typing import Optional
 
 from mpflash.errors import MPFlashError
-from mpflash.vendor.versions import clean_version
+from mpflash.vendor.versions import clean_version, get_stable_mp_version
 
 ###############################################################################################
 HERE = Path(__file__).parent
@@ -48,8 +48,8 @@ def _find_board_id_by_description(
 
     # filter for matching version
     if version == "preview":
-        # TODO: match last stable
-        version = "v1.22.2"
+        # match last stable
+        version = get_stable_mp_version()
     version_matches = [b for b in info if b["version"].startswith(version)]
     if not version_matches:
         raise MPFlashError(f"No board info found for version {version}")

src/mpflash/mpflash/worklist.py

@@ -9,7 +9,7 @@
 from .config import config
 from .downloaded import find_downloaded_firmware
 from .list import show_mcus
-from .mpboard_id import find_stored_board
+from .mpboard_id import find_known_board
 from .mpremoteboard import MPRemoteBoard
 
 # #########################################################################################################
@@ -131,7 +131,7 @@ def manual_worklist(
     # TODO : Find a way to avoid needing to specify the port
     # Lookup the matching port and cpu in board_info based in the board name
     try:
-        info = find_stored_board(board)
+        info = find_known_board(board)
         mcu.port = info["port"]
         # need the CPU type for the esptool
         mcu.cpu = info["cpu"]