Skip to content

Commit

Permalink
Add query subcommand
Browse files Browse the repository at this point in the history
  • Loading branch information
tekktrik committed Feb 26, 2024
1 parent c66f487 commit 9c0138a
Show file tree
Hide file tree
Showing 5 changed files with 165 additions and 4 deletions.
1 change: 1 addition & 0 deletions circfirm/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

# Files
SETTINGS_FILE = setup_file(APP_DIR, "settings.yaml")
UF2_BOARD_LIST = setup_file(APP_DIR, "boards.txt")

UF2INFO_FILE = "info_uf2.txt"
BOOTOUT_FILE = "boot_out.txt"
66 changes: 62 additions & 4 deletions circfirm/backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,15 @@
import re
from typing import Dict, List, Optional, Set, Tuple

import boto3
import botocore
import botocore.client
import github
import github.GitTree
import packaging.version
import psutil
import requests
from mypy_boto3_s3 import S3ServiceResource

import circfirm
import circfirm.startup
Expand Down Expand Up @@ -48,15 +54,31 @@ 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"


def _find_device(filename: str) -> Optional[str]:
"""Find a specific connected device."""
Expand Down Expand Up @@ -170,3 +192,39 @@ 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_board_list() -> List[str]:
"""Get a list of CircuitPython boards."""
boards = set()
gh = github.Github()
repo = gh.get_repo("adafruit/circuitpython")
tree = repo.get_git_tree("main", True)
for element in tree.tree:
result = re.match(BOARDS_REGEX, element.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)
98 changes: 98 additions & 0 deletions circfirm/cli/query.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
# SPDX-FileCopyrightText: 2024 Alec Delaney, for Adafruit Industries
#
# SPDX-License-Identifier: MIT

"""CLI functionality for the query subcommand.
Author(s): Alec Delaney
"""

import re

import click
import github
import packaging.version

import circfirm
import circfirm.backend
import circfirm.cli


@click.group()
def cli():
"""Query things like latest versions and board lists."""


def query_sync() -> None:
"""Sync CircuitPython board information."""
click.echo("Boards list will now be synchronized with the git repository.")
click.echo(
"Please note that this operation can only be performed 60 times per hour due to GitHub rate limiting."
)
try:
boards = circfirm.cli.announce_and_await(
"Updating boards list", circfirm.backend.get_board_list, use_spinner=True
)
except github.GithubException as err:
raise click.ClickException(err.data["message"])
with open(circfirm.UF2_BOARD_LIST, mode="w", encoding="utf-8") as boardfile:
for board in boards:
boardfile.write(f"{board}\n")


@cli.command(name="boards")
@click.option(
"-r", "--regex", default=None, help="Regex pattern to use for board names"
)
@click.option("-s", "--sync", is_flag=True, default=False, help="Sync the boards list")
def query_boards(regex: str, sync: bool) -> None:
"""Query the synchronized CircuitPython board list."""
if sync and regex:
raise click.ClickException("Cannot use --regex and --sync simultaneously.")

if sync:
query_sync()
return

if regex is None:
regex = ".*"

with open(circfirm.UF2_BOARD_LIST, encoding="utf-8") as boardfile:
for board in boardfile:
board_name = board.strip()
result = re.match(regex, board_name)
if result:
click.echo(board_name)


@cli.command(name="versions")
@click.argument("board")
@click.option("-l", "--language", default="en_US", help="CircuitPython language/locale")
@click.option("-r", "--regex", default=".*", help="Regex pattern to use for versions")
def query_versions(board: str, language: str, regex: str) -> None:
"""Query the CircuitPython versions available for a board."""
versions = circfirm.backend.get_board_versions(board, language, regex=regex)
for version in reversed(versions):
click.echo(version)


@cli.command(name="latest")
@click.argument("board")
@click.option("-l", "--language", default="en_US", help="CircuitPython language/locale")
@click.option(
"-p",
"--pre-release",
is_flag=True,
default=False,
help="Consider pre-release versions",
)
def query_latest(board: str, language: str, pre_release: bool) -> None:
"""Query the latest CircuitPython versions available for a board."""
versions = circfirm.backend.get_board_versions(board, language)
if not pre_release:
versions = [
version
for version in versions
if not packaging.version.Version(version).is_prerelease
]
click.echo(versions[0])
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -78,3 +78,6 @@ source = [
[tool.ruff.lint]
select = ["D", "PL", "UP", "I"]
ignore = ["D213", "D203"]

[tool.pytest.ini_options]
filterwarnings = ["ignore:datetime.datetime.utcfromtimestamp():DeprecationWarning"]
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ click~=8.0
click-spinner~=0.1
packaging~=23.2
psutil~=5.9
pygithub~=2.2
pyyaml~=6.0
requests~=2.31
boto3-stubs[essential]~=1.34

0 comments on commit 9c0138a

Please sign in to comment.