Skip to content

Commit

Permalink
Refactor blueos_repository to use new modules
Browse files Browse the repository at this point in the history 
* Change current blueos_repository to use new dedicated modules and to
  generate all extensions in parallel as well as adding the manifest.log
  file in the end of the generation process
  • Loading branch information
@JoaoMario109
JoaoMario109 committed Apr 26, 2024
1 parent b0570c0 commit 6be4d16
Showing 1 changed file with 110 additions and 209 deletions.
319 changes: 110 additions & 209 deletions blueos_repository/consolidate.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,259 +2,160 @@
import asyncio
import dataclasses
import json
from enum import Enum
from pathlib import Path
from typing import Any, AsyncIterable, Dict, List, Optional, Union
from typing import AsyncIterable, Dict, Optional, Tuple

import aiohttp
import json5
import semver
from registry import Registry
from docker.registry import DockerRegistry
from extension.extension import Extension
from extension.models import ExtensionMetadata, ExtensionVersion
from logger import Logger
from utils import EnhancedJSONEncoder

REPO_ROOT = "https://raw.githubusercontent.com/bluerobotics/BlueOS-Extensions-Repository/master/repos"
# This is the name manifest file that will be generated
MANIFEST_FILE = "manifest.json"

# This is the name of the log file that will be generated
MANIFEST_LOG = "manifest.log"

class StrEnum(str, Enum):
"""Temporary filler until Python 3.11 available."""

def __str__(self) -> str:
return self.value # type: ignore


class EnhancedJSONEncoder(json.JSONEncoder):
"""
Custom json encoder for dataclasses,
see https://docs.python.org/3/library/json.html#json.JSONEncoder.default
Returns a serializable type
"""

def default(self, o: Any) -> Union[Any, Dict[str, Any]]:
if dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
return super().default(o)
# This is the root URL for the public repository
REPO_ROOT_URL = "https://raw.githubusercontent.com/bluerobotics/BlueOS-Extensions-Repository/master/repos"


@dataclasses.dataclass
class Author:
name: str
email: str

@staticmethod
def from_json(json_dict: Dict[str, str]) -> "Author":
return Author(name=json_dict["name"], email=json_dict["email"])


@dataclasses.dataclass
class Platform:
architecture: str
variant: Optional[str] = None
# pylint: disable=invalid-name
os: Optional[str] = None
class RepositoryEntry(ExtensionMetadata):
"""
Represents a repository entry in the manifest output
Attributes:
versions (Dict[str, Version]): Available extension versions.
"""

@dataclasses.dataclass
class Image:
digest: Optional[str]
expanded_size: int
platform: Platform
versions: Dict[str, ExtensionVersion] = dataclasses.field(default_factory=dict)


@dataclasses.dataclass
class Company:
name: str
about: Optional[str]
email: Optional[str]
class Consolidator:
"""
This class is used to consolidate the BlueOS extensions repository generating
a manifest file with all the extensions available and their versions.
"""

@staticmethod
def from_json(json_dict: Dict[str, str]) -> "Company":
return Company(name=json_dict["name"], email=json_dict.get("email", None), about=json_dict.get("about", None))

def repo_folder() -> Path:
return Path(__file__).parent.parent.joinpath("repos")

class ExtensionType(StrEnum):
DEVICE_INTEGRATION = "device-integration"
EXAMPLE = "example"
THEME = "theme"
OTHER = "other"
TOOL = "tool"

def fetch_remote_extension_logos(
self,
identifier: str,
repository: Path,
repositories: Path
) -> Tuple[Optional[str], Optional[str]]:
"""
Fetch the remote extension and company logos for a given repository.
Args:
identifier (str): Extension identifier.
repository (Path): Path to the repository folder.
repositories (Path): Path to the repositories folder.
Returns:
Tuple[Optional[str], Optional[str]]: Remote extension and company logos URLs.
"""

company_logo = (repository / "../../company_logo.png").resolve().relative_to(repositories.resolve())
extension_logo_file = (repository / "../extension_logo.png").resolve()

if extension_logo_file.exists():
extension_logo = extension_logo_file.resolve().relative_to(repositories.resolve())
else:
Logger.warning(
identifier,
f"Extension logo not found for {identifier}, trying to use company logo as alternative"
)
extension_logo = company_logo

# pylint: disable=too-many-instance-attributes
@dataclasses.dataclass
class Version:
permissions: Optional[Dict[str, Any]]
requirements: Optional[str]
tag: Optional[str]
website: str
authors: List[Author]
docs: Optional[str]
readme: Optional[str]
company: Optional[Company]
support: Optional[str]
type: ExtensionType
filter_tags: List[str]
extra_links: Dict[str, str]
images: List[Image]
remote_extension_logo = f"{REPO_ROOT_URL}/{extension_logo}" if extension_logo else None
remote_company_logo = f"{REPO_ROOT_URL}/{company_logo}" if company_logo else None

@staticmethod
def validate_filter_tags(tags: List[str]) -> List[str]:
"""Returns a list of up to 10 lower-case alpha-numeric tags (dashes allowed)."""
return [tag.lower() for tag in tags if tag.replace("-", "").isalnum()][:10]
if not remote_company_logo or not remote_extension_logo:
Logger.warning(identifier, f"Unable to find extension or company logo for {identifier}")

return remote_extension_logo, remote_company_logo

@dataclasses.dataclass
class RepositoryEntry:
identifier: str
name: str
description: str
docker: str
website: str
versions: Dict[str, Version]
extension_logo: Optional[str]
company_logo: Optional[str]

async def fetch_extensions_metadata(self) -> AsyncIterable[ExtensionMetadata]:
"""
Fetch the metadata for all the extensions in the repository.
class Consolidator:
registry = Registry()
consolidated_data: List[RepositoryEntry] = []
Returns:
List[ExtensionMetadata]: List of all the extensions metadata.
"""

@staticmethod
def repo_folder() -> Path:
return Path(__file__).parent.parent.joinpath("repos")

@staticmethod
async def fetch_readme(url: str) -> str:
if not url.startswith("http"):
print(f"Invalid Readme url: {url}")
return "Readme not provided."
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
if resp.status != 200:
print(f"Error status {resp.status}")
raise Exception(f"Could not get readme {url}: status: {resp.status}")
if resp.content_type != "text/plain":
raise Exception(f"bad response type for readme: {resp.content_type}, expected text/plain")
return await resp.text()

async def all_repositories(self) -> AsyncIterable[RepositoryEntry]:
repos = self.repo_folder()

for repo in repos.glob("**/metadata.json"):
with open(repo, "r", encoding="utf-8") as individual_file:
with open(repo, "r", encoding="utf-8") as metadata_file:
company, extension_name = repo.as_posix().split("/")[-3:-1]
identifier = ".".join([company, extension_name])

Logger.info(identifier, f"Starting metadata processing for {identifier}")

try:
data = json5.load(individual_file)
except Exception as exc:
raise Exception(f"Unable to parse file {repo}") from exc
company_logo = (repo / "../../company_logo.png").resolve().relative_to(repos.resolve())
extension_logo_file = (repo / "../extension_logo.png").resolve()
if extension_logo_file.exists():
extension_logo = extension_logo_file.resolve().relative_to(repos.resolve())
else:
extension_logo = company_logo
data = json5.load(metadata_file)
except Exception as error:
Logger.error(identifier, f"Skipping {identifier}, unable to parse metadata file, error: {error}")
continue

extension_logo, company_logo = self.fetch_remote_extension_logos(identifier, repo, repos)

try:
new_repo = RepositoryEntry(
metadata = ExtensionMetadata(
identifier=identifier,
name=data["name"],
docker=data["docker"],
description=data["description"],
website=data["website"],
extension_logo=f"{REPO_ROOT}/{extension_logo}" if extension_logo else None,
versions={},
company_logo=f"{REPO_ROOT}/{company_logo}" if company_logo else None,
extension_logo=extension_logo,
company_logo=company_logo,
)
yield new_repo

Logger.info(identifier, f"Finished metadata processing for {identifier}")
yield metadata
except Exception as error:
raise Exception(f"unable to read file {repo}: {error}") from error
Logger.error(identifier, f"Skipping {identifier}, invalid metadata file, error: {error}")
continue


async def run(self) -> None:
preview = DockerRegistry.from_preview()

@staticmethod
def valid_semver(string: str) -> Optional[semver.VersionInfo]:
# We want to allow versions to be prefixed with a 'v'.
# This is up for discussion
if string.startswith("v"):
string = string[1:]
try:
return semver.VersionInfo.parse(string)
except ValueError:
return None # not valid

def extract_images_from_tag(self, tag: Any) -> List[Image]:
active_images = [
image
for image in tag["images"]
if (image["status"] == "active" and image["architecture"] != "unknown" and image["os"] != "unknown")
]
Logger.start_docker_rate_limit(await preview.get_rate_limit())
except Exception as error:
print(f"Unable to fetch initial docker rate limit, error: {error}")

extensions = [Extension(metadata) async for metadata in self.fetch_extensions_metadata()]

await asyncio.gather(*(ext.inflate() for ext in extensions))

images = [
Image(
digest=image.get("digest", None),
expanded_size=image["size"],
platform=Platform(
architecture=image["architecture"],
variant=image.get("variant", None),
os=image.get("os", None),
),
consolidated_data = [
RepositoryEntry(
**dataclasses.asdict(ext.metadata),
versions=ext.versions
)
for image in active_images
for ext in extensions if ext.versions
]
return images

# pylint: disable=too-many-locals
async def run(self) -> None:
async for repository in self.all_repositories():
for tag in await self.registry.fetch_remote_tags(repository.docker):
tag_name = tag["name"]
print(tag_name)
try:
if not self.valid_semver(tag_name):
print(f"{tag_name} is not valid SemVer, ignoring it...")
continue
raw_labels = await self.registry.fetch_labels(f"{repository.docker}:{tag_name}")
permissions = raw_labels.get("permissions", None)
links = json5.loads(raw_labels.get("links", "{}"))
website = links.pop("website", raw_labels.get("website", None))
authors = json5.loads(raw_labels.get("authors", "[]"))
# documentation is just a URL for a link, but the old format had it as its own label
docs = links.pop("docs", links.pop("documentation", raw_labels.get("docs", None)))
readme = raw_labels.get("readme", None)
if readme is not None:
readme = readme.replace(r"{tag}", tag_name)
try:
readme = await self.fetch_readme(readme)
except Exception as error: # pylint: disable=broad-except
readme = str(error)
company_raw = raw_labels.get("company", None)
company = Company.from_json(json5.loads(company_raw)) if company_raw is not None else None
support = links.pop("support", raw_labels.get("support", None))
type_ = raw_labels.get("type", ExtensionType.OTHER)
filter_tags = json5.loads(raw_labels.get("tags", "[]"))

new_version = Version(
permissions=json5.loads(permissions) if permissions else None,
website=website,
authors=authors,
docs=json5.loads(docs) if docs else None,
readme=readme,
company=company,
support=support,
extra_links=links,
type=type_,
filter_tags=Version.validate_filter_tags(filter_tags),
requirements=raw_labels.get("requirements", None),
tag=tag_name,
images=self.extract_images_from_tag(tag),
)
repository.versions[tag_name] = new_version
except KeyError as error:
raise Exception(f"unable to parse repository {repository}: {error}") from error
# sort the versions, with the highest version first
repository.versions = dict(
sorted(repository.versions.items(), key=lambda i: self.valid_semver(i[0]), reverse=True) # type: ignore
)
if repository.versions: # only include if there's at least one valid version
self.consolidated_data.append(repository)
try:
Logger.final_docker_rate_limit(await preview.get_rate_limit())
except Exception as error:
print(f"Unable to fetch final docker rate limit, error: {error}")

with open(MANIFEST_FILE, "w", encoding="utf-8") as manifest_file:
manifest_file.write(json.dumps(consolidated_data, indent=4, sort_keys=True, cls=EnhancedJSONEncoder))

with open("manifest.json", "w", encoding="utf-8") as manifest_file:
manifest_file.write(json.dumps(self.consolidated_data, indent=4, cls=EnhancedJSONEncoder))
Logger.dump(MANIFEST_LOG)


consolidator = Consolidator()
Expand Down

0 comments on commit 6be4d16

Please sign in to comment.