Merge pull request #55 from tekktrik/dev/query

Add querry and config commands

.github/workflows/push.yml

@@ -6,6 +6,8 @@ name: Build CI
 
 on: ["push", "pull_request"]
 
+permissions: read-all
+
 jobs:
   build:
     name: Run build CI
@@ -45,6 +47,8 @@ jobs:
       - name: Run tests
         run: |
           make test-run
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       - name: Clean up after tests
         run: |
           make test-clean

.gitignore

@@ -141,3 +141,4 @@ testmount/
 
 # Test related
 tests/backup/
+tests/sandbox/

Makefile

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: MIT
 
-
 .PHONY: lint
 lint:
 	@pre-commit run ruff --all-files
@@ -35,6 +34,7 @@ else
 	@echo "Current OS not supported"
 	@exit 1
 endif
+	-@git clone https://github.com/adafruit/circuitpython tests/sandbox/circuitpython --depth 1
 
 .PHONY: test
 test:
@@ -53,11 +53,14 @@ test-clean:
 ifeq "$(OS)" "Windows_NT"
 	-@subst T: /d
 	-@python scripts/rmdir.py testmount
+	-@python scripts/rmdir.py tests/sandbox/circuitpython
 else ifeq "$(shell uname -s)" "Linux"
 	-@sudo umount testmount
 	-@sudo rm -rf testmount
 	-@rm testfs -f
+	-@rm -rf tests/sandbox/circuitpython
 else
 	-@hdiutil detach /Volumes/TESTMOUNT
 	-@rm testfs.dmg -f
+	-@rm -rf tests/sandbox/circuitpython
 endif

circfirm/__init__.py

@@ -7,14 +7,27 @@
 Author(s): Alec Delaney
 """
 
-from circfirm.startup import setup_app_dir, setup_file, setup_folder
+import os
+
+from circfirm.startup import (
+    specify_app_dir,
+    specify_file,
+    specify_folder,
+    specify_template,
+)
 
 # Folders
-APP_DIR = setup_app_dir("circfirm")
-UF2_ARCHIVE = setup_folder(APP_DIR, "archive")
+APP_DIR = specify_app_dir("circfirm")
+UF2_ARCHIVE = specify_folder(APP_DIR, "archive")
 
 # Files
-SETTINGS_FILE = setup_file(APP_DIR, "settings.yaml")
+_SETTINGS_FILE_SRC = os.path.abspath(
+    os.path.join(__file__, "..", "templates", "settings.yaml")
+)
+SETTINGS_FILE = specify_template(
+    _SETTINGS_FILE_SRC, os.path.join(APP_DIR, "settings.yaml")
+)
+UF2_BOARD_LIST = specify_file(APP_DIR, "boards.txt")
 
 UF2INFO_FILE = "info_uf2.txt"
 BOOTOUT_FILE = "boot_out.txt"

circfirm/backend.py

@@ -7,15 +7,20 @@
 Author(s): Alec Delaney
 """
 
+import datetime
 import enum
 import os
 import pathlib
 import re
-from typing import Dict, List, Optional, Set, Tuple
+from typing import Dict, List, Optional, Set, Tuple, TypedDict
 
+import boto3
+import botocore
+import botocore.client
 import packaging.version
 import psutil
 import requests
+from mypy_boto3_s3 import S3ServiceResource
 
 import circfirm
 import circfirm.startup
@@ -48,15 +53,57 @@ class Language(enum.Enum):
 
 _ALL_LANGAGES = [language.value for language in Language]
 _ALL_LANGUAGES_REGEX = "|".join(_ALL_LANGAGES)
-FIRMWARE_REGEX = "-".join(
+_VALID_VERSIONS_CAPTURE = r"(\d+\.\d+\.\d+(?:-(?:\balpha\b|\bbeta\b)\.\d+)*)"
+FIRMWARE_REGEX_PATTERN = "-".join(
     [
-        r"adafruit-circuitpython-(.*)",
-        f"({_ALL_LANGUAGES_REGEX})",
-        r"(\d+\.\d+\.\d+(?:-(?:\balpha\b|\bbeta\b)\.\d+)*)\.uf2",
+        r"adafruit-circuitpython",
+        r"[board]",
+        r"[language]",
+        r"[version]\.uf2",
     ]
 )
+FIRMWARE_REGEX = (
+    FIRMWARE_REGEX_PATTERN.replace(r"[board]", r"(.*)")
+    .replace(r"[language]", f"({_ALL_LANGUAGES_REGEX})")
+    .replace(r"[version]", _VALID_VERSIONS_CAPTURE)
+)
+
 BOARD_ID_REGEX = r"Board ID:\s*(.*)"
 
+S3_CONFIG = botocore.client.Config(signature_version=botocore.UNSIGNED)
+S3_RESOURCE: S3ServiceResource = boto3.resource("s3", config=S3_CONFIG)
+BUCKET_NAME = "adafruit-circuit-python"
+BUCKET = S3_RESOURCE.Bucket(BUCKET_NAME)
+
+BOARDS_REGEX = r"ports/.+/boards/([^/]+)"
+BOARDS_REGEX_PATTERN2 = r"bin/([board_pattern])/en_US/.*\.uf2"
+
+_BASE_REQUESTS_HEADERS = {
+    "Accept": "application/vnd.github+json",
+    "X-GitHub-Api-Version": "2022-11-28",
+}
+
+
+class RateLimit(TypedDict):
+    """Format of a rate limit dictionary."""
+
+    limit: int
+    remaining: int
+    reset: int
+    used: int
+    resource: str
+
+
+class GitTreeItem(TypedDict):
+    """Format of a git tree item dictionary."""
+
+    path: str
+    mode: str
+    type: str
+    size: int
+    sha: str
+    url: str
+
 
 def _find_device(filename: str) -> Optional[str]:
     """Find a specific connected device."""
@@ -102,7 +149,7 @@ def download_uf2(board: str, version: str, language: str = "en_US") -> None:
     if response.status_code != SUCCESS:
         if not list(uf2_file.parent.glob("*")):
             uf2_file.parent.rmdir()
-        raise ConnectionError(f"Could not download spectified UF2 file:\n{url}")
+        raise ConnectionError(f"Could not download the specified UF2 file:\n{url}")
 
     with open(uf2_file, mode="wb") as uf2file:
         uf2file.write(response.content)
@@ -170,3 +217,65 @@ def get_sorted_boards(board: Optional[str]) -> Dict[str, Dict[str, Set[str]]]:
             sorted_versions[sorted_version] = versions[sorted_version]
         boards[board_folder] = sorted_versions
     return boards
+
+
+def get_rate_limit() -> Tuple[int, int, datetime.datetime]:
+    """Get the rate limit for the GitHub REST endpoint."""
+    response = requests.get(
+        url="https://api.github.com/rate_limit",
+        headers=_BASE_REQUESTS_HEADERS,
+    )
+    limit_info: RateLimit = response.json()["rate"]
+    available: int = limit_info["remaining"]
+    total: int = limit_info["limit"]
+    reset_time = datetime.datetime.fromtimestamp(limit_info["reset"])
+    return available, total, reset_time
+
+
+def get_board_list(token: str) -> List[str]:
+    """Get a list of CircuitPython boards."""
+    boards = set()
+    headers = _BASE_REQUESTS_HEADERS.copy()
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    response = requests.get(
+        url="https://api.github.com/repos/adafruit/circuitpython/git/trees/main",
+        params={
+            "recursive": True,
+        },
+        headers=headers,
+    )
+    try:
+        tree_items: List[GitTreeItem] = response.json()["tree"]
+    except KeyError as err:
+        raise ValueError("Could not parse JSON response, check token") from err
+    for tree_item in tree_items:
+        if tree_item["type"] != "tree":
+            continue
+        result = re.match(BOARDS_REGEX, tree_item["path"])
+        if result:
+            boards.add(result[1])
+    return sorted(boards)
+
+
+def get_board_versions(
+    board: str, language: str = "en_US", *, regex: Optional[str] = None
+) -> List[str]:
+    """Get a list of CircuitPython versions for a given board."""
+    prefix = f"bin/{board}/{language}"
+    firmware_regex = FIRMWARE_REGEX_PATTERN.replace(r"[board]", board).replace(
+        r"[language]", language
+    )
+    version_regex = f"({regex})" if regex else _VALID_VERSIONS_CAPTURE
+    firmware_regex = firmware_regex.replace(r"[version]", version_regex)
+    s3_objects = BUCKET.objects.filter(Prefix=prefix)
+    versions = set()
+    for s3_object in s3_objects:
+        result = re.match(f"{prefix}/{firmware_regex}", s3_object.key)
+        if result:
+            try:
+                _ = packaging.version.Version(result[1])
+                versions.add(result[1])
+            except packaging.version.InvalidVersion:
+                pass
+    return sorted(versions, key=packaging.version.Version, reverse=True)

circfirm/cli/__init__.py

@@ -13,14 +13,18 @@
 import shutil
 import sys
 import time
-from typing import Any, Callable, Dict, Iterable, Optional
+from typing import Any, Callable, Dict, Iterable, Optional, TypeVar
 
 import click
+import click_spinner
+import yaml
 
 import circfirm
 import circfirm.backend
 import circfirm.startup
 
+_T = TypeVar("_T")
+
 
 @click.group()
 @click.version_option(package_name="circfirm")
@@ -29,20 +33,47 @@ def cli() -> None:
     circfirm.startup.ensure_app_setup()
 
 
+def maybe_support(msg: str) -> None:
+    """Output supporting text based on the configurable settings."""
+    settings = get_settings()
+    do_output: bool = not settings["output"]["supporting"]["silence"]
+    if do_output:
+        click.echo(msg)
+
+
 def announce_and_await(
     msg: str,
-    func: Callable,
+    func: Callable[..., _T],
     args: Iterable = (),
     kwargs: Optional[Dict[str, Any]] = None,
-) -> Any:
+    *,
+    use_spinner: bool = True,
+) -> _T:
     """Announce an action to be performed, do it, then announce its completion."""
     if kwargs is None:
         kwargs = {}
     fmt_msg = f"{msg}..."
+    spinner = click_spinner.spinner()
     click.echo(fmt_msg, nl=False)
-    resp = func(*args, **kwargs)
-    click.echo(" done")
-    return resp
+    if use_spinner:
+        spinner.start()
+    try:
+        try:
+            resp = func(*args, **kwargs)
+        finally:
+            if use_spinner:
+                spinner.stop()
+        click.echo(" done")
+        return resp
+    except BaseException as err:
+        click.echo(" failed")
+        raise err
+
+
+def get_settings() -> Dict[str, Any]:
+    """Get the contents of the settings file."""
+    with open(circfirm.SETTINGS_FILE, encoding="utf-8") as yamlfile:
+        return yaml.safe_load(yamlfile)
 
 
 def load_subcmd_folder(path: str, super_import_name: str) -> None:
@@ -110,7 +141,6 @@ def install(version: str, language: str, board: Optional[str]) -> None:
                 args=(board, version, language),
             )
         except ConnectionError as err:
-            click.echo(" failed")  # Mark as failed
             click.echo(f"Error: {err.args[0]}")
             sys.exit(4)
     else:

circfirm/cli/cache.py

@@ -43,7 +43,7 @@ def clear(
     glob_pattern = "*-*" if board is None else f"*-{board}"
     language_pattern = "-*" if language is None else f"-{language}"
     glob_pattern += language_pattern
-    version_pattern = "-*" if version is None else f"-{version}*"
+    version_pattern = "-*" if version is None else f"-{version}.uf2"
     glob_pattern += version_pattern
     matching_files = pathlib.Path(circfirm.UF2_ARCHIVE).rglob(glob_pattern)
     for matching_file in matching_files:
@@ -64,11 +64,11 @@ def cache_list(board: Optional[str]) -> None:
     board_list = os.listdir(circfirm.UF2_ARCHIVE)
 
     if not board_list:
-        click.echo("Versions have not been cached yet for any boards.")
+        circfirm.cli.maybe_support("Versions have not been cached yet for any boards.")
         sys.exit(0)
 
     if board is not None and board not in board_list:
-        click.echo(f"No versions for board '{board}' are not cached.")
+        circfirm.cli.maybe_support(f"No versions for board '{board}' are not cached.")
         sys.exit(0)
 
     specified_board = board if board is not None else None
@@ -94,5 +94,4 @@ def cache_save(board: str, version: str, language: str) -> None:
             args=(board, version, language),
         )
     except ConnectionError as err:
-        click.echo(" failed")  # Mark as failed
         raise click.exceptions.ClickException(err.args[0])