246 lines
7.0 KiB
Python
246 lines
7.0 KiB
Python
"""
|
|
CircuitPack extractor - install CircuitPack packages
|
|
"""
|
|
|
|
import os
|
|
import requests
|
|
|
|
try:
|
|
from typing import Sequence, BinaryIO, Iterable, Union
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
def _mkdir_p(parts: Sequence[str]) -> None:
|
|
whole = "/".join(parts)
|
|
|
|
# This function is based in part on code from CPython's `pathlib` module,
|
|
# specifically the `pathlib2` backport, the relevant portion of which does
|
|
# not differ from CPython's tree as of the time of writing.
|
|
#
|
|
# The code used is available at the following URL:
|
|
#
|
|
# https://github.com/jazzband/pathlib2/blob/0ac11b8d697069abea58a8188061f1a1870977af/src/pathlib2/__init__.py#L1124
|
|
#
|
|
# That file includes the following copyright notice:
|
|
#
|
|
# Copyright (c) 2001-2022 Python Software Foundation; All Rights Reserved
|
|
# Copyright (c) 2014-2022 Matthias C. M. Troffaes and contributors
|
|
# Copyright (c) 2012-2014 Antoine Pitrou and contributors
|
|
#
|
|
# Distributed under the terms of the MIT License.
|
|
#
|
|
# (end copyright notice)
|
|
#
|
|
# I determined the logic and magic numbers to use for directory checking
|
|
# from CPython's `stat` module, but did not directly use a significant
|
|
# amount of its code. `stat`'s source code can be found here:
|
|
#
|
|
# https://github.com/python/cpython/blob/717d2bd1d95879992d30d417399b1ec03ee05028/Lib/stat.py
|
|
|
|
try:
|
|
os.mkdir(whole)
|
|
except FileNotFoundError:
|
|
_mkdir_p(parts[:-1])
|
|
os.mkdir(whole)
|
|
except OSError:
|
|
if not os.stat(whole).st_mode & 0o170000 == 0o040000:
|
|
raise
|
|
|
|
# (end code derived from `pathlib`)
|
|
|
|
|
|
class InvalidFileError(BaseException):
|
|
"""
|
|
CircuitPack package file is invalid.
|
|
"""
|
|
|
|
|
|
class _ResponseFile: # pylint: disable=too-few-public-methods
|
|
def __init__(
|
|
self, response: requests.Response, chunk_size: int = 128
|
|
) -> None:
|
|
self.iter = response.iter_content(chunk_size)
|
|
self.buffer = b""
|
|
|
|
def read( # pylint: disable=missing-function-docstring
|
|
self, length: int
|
|
) -> bytes:
|
|
try:
|
|
while len(self.buffer) < length:
|
|
self.buffer += next(self.iter)
|
|
except StopIteration:
|
|
pass
|
|
|
|
response = self.buffer[:length]
|
|
self.buffer = self.buffer[length:]
|
|
|
|
return response
|
|
|
|
def readline(self) -> bytes: # pylint: disable=missing-function-docstring
|
|
try:
|
|
while not b"\n" in self.buffer:
|
|
self.buffer += next(self.iter)
|
|
except StopIteration:
|
|
pass
|
|
|
|
if b"\n" == self.buffer:
|
|
response, self.buffer = self.buffer.split(b"\n", 1)
|
|
return response
|
|
else:
|
|
response = self.buffer
|
|
self.buffer = b""
|
|
return response
|
|
|
|
|
|
def unpack(
|
|
package: Union[BinaryIO, _ResponseFile], target: str
|
|
) -> Iterable[str]:
|
|
"""
|
|
Unpack the given file-like object `package` into the directory `target`,
|
|
which may, but is not required to, already exist. Return the unpacked
|
|
paths, relative to `target`.
|
|
"""
|
|
|
|
if not target.endswith("/"):
|
|
target += "/"
|
|
|
|
if package.read(8) != b"CPAv001\n":
|
|
raise InvalidFileError("wrong header")
|
|
|
|
try:
|
|
length = int(package.read(8).decode("utf-8"))
|
|
except ValueError as exception:
|
|
raise InvalidFileError("invalid index length") from exception
|
|
|
|
raw_index_bytes = package.read(length)
|
|
if len(raw_index_bytes) < length:
|
|
raise InvalidFileError("file cut short (incomplete index)")
|
|
|
|
try:
|
|
raw_index = raw_index_bytes.decode("utf-8")
|
|
except UnicodeDecodeError as exception:
|
|
raise InvalidFileError("invalid character in index") from exception
|
|
|
|
index_split = [line.split(" ") for line in raw_index.split("\n")[:-1]]
|
|
|
|
try:
|
|
index = {file_name: int(length) for file_name, length in index_split}
|
|
except ValueError as exception:
|
|
raise InvalidFileError(
|
|
"incorrect number of fields or invalid number in index"
|
|
) from exception
|
|
|
|
for file_name, length in index.items():
|
|
path = target + file_name
|
|
_mkdir_p(path.split("/")[:-1])
|
|
with open(path, "wb+") as file:
|
|
if file.write(package.read(length)) < length:
|
|
raise InvalidFileError(
|
|
"file cut short (or, possibly, a write error)"
|
|
)
|
|
|
|
return index.keys()
|
|
|
|
|
|
def unpack_url(url: str, target: str) -> Iterable[str]:
|
|
"""
|
|
Download the package at `url` and unpack it into the directory `target`,
|
|
which may, but is not required to, already exist. Return the unpacked
|
|
paths, relative to `target`.
|
|
"""
|
|
|
|
return unpack(_ResponseFile(requests.get(url)), target)
|
|
|
|
|
|
def install(
|
|
name: str, version_hash: str, repo: str, target: str, records: str
|
|
) -> None:
|
|
"""
|
|
Download version `version_hash` of package `name` from `repo`, unpack it
|
|
into the directory `target`, and create a record file in `records`.
|
|
"""
|
|
|
|
if not repo.endswith("/"):
|
|
repo = repo + "/"
|
|
|
|
file_names = unpack_url(f"{repo}{name}.{version_hash}.cpa", target)
|
|
|
|
with open(f"{records}/{name}", "w+", encoding="utf-8") as record:
|
|
record.write(f"{version_hash}\n")
|
|
|
|
for file_name in file_names:
|
|
record.write(f"{file_name}\n")
|
|
|
|
|
|
def uninstall(name: str, target: str, records: str) -> None:
|
|
"""
|
|
Uninstall package `name` from `target` using the information stored in
|
|
`records`.
|
|
"""
|
|
|
|
with open(f"{records}/{name}", "r", encoding="utf-8") as record:
|
|
record.readline() # Skip over version hash
|
|
|
|
while True:
|
|
file_name = record.readline().strip()
|
|
|
|
if file_name:
|
|
os.remove(f"{target}/{file_name}")
|
|
else:
|
|
break
|
|
|
|
os.remove(f"{records}/{name}")
|
|
|
|
|
|
def _vacuum(root: str, delete_this: bool = False) -> None:
|
|
try:
|
|
for item in os.listdir(root):
|
|
_vacuum(f"{root}/{item}", True)
|
|
except NotADirectoryError:
|
|
pass
|
|
|
|
if delete_this:
|
|
try:
|
|
os.rmdir(root)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _get_status(records: str) -> dict[str, str]:
|
|
status = {}
|
|
|
|
for name in os.listdir(records):
|
|
with open(f"{records}/{name}", encoding="utf-8") as f:
|
|
status[name] = f.readline().strip()
|
|
|
|
return status
|
|
|
|
|
|
def install_spec(
|
|
spec_text: str, target: str, records: str, paths: Iterable[str] = ("lib",)
|
|
) -> None:
|
|
spec = {}
|
|
|
|
for full_line in spec_text.split("\n"):
|
|
line = full_line.strip()
|
|
if line:
|
|
name, version_hash, repo = line.split(",")
|
|
spec[name] = (version_hash, repo)
|
|
|
|
status = _get_status(records)
|
|
|
|
for name, version in status.items():
|
|
if not name in spec or spec[name][0] != version:
|
|
uninstall(name, target, records)
|
|
|
|
for path in paths:
|
|
_vacuum(f"{target}/{path}")
|
|
status = _get_status(records)
|
|
|
|
for name, version_repo in spec.items():
|
|
version, repo = version_repo
|
|
|
|
if not name in status:
|
|
install(name, version, repo, target, records)
|