diff --git a/.gitignore b/.gitignore index 3b557d6..f07fca1 100644 --- a/.gitignore +++ b/.gitignore @@ -7,10 +7,12 @@ venv *.egg-info .DS_Store .vscode + dist secrets.yaml secrets.yml settings.json inkBoard/platforms/*/ inkBoard/integrations/*/ +inkBoard/packaging/*.json inkBoarddesigner \ No newline at end of file diff --git a/inkBoard/packaging.py b/inkBoard/_packaging.py similarity index 99% rename from inkBoard/packaging.py rename to inkBoard/_packaging.py index 4d181de..ac3feec 100644 --- a/inkBoard/packaging.py +++ b/inkBoard/_packaging.py @@ -1,3 +1,5 @@ +"backup version of packaging.py for porting" + "Handles inkBoard packages, both creating and installing them." import asyncio @@ -25,6 +27,10 @@ import PythonScreenStackManager as PSSM +from .constants import ( + ZIP_COMPRESSION, + ZIP_COMPRESSION_LEVEL, + ) if TYPE_CHECKING: from inkBoard import CORE as CORE @@ -38,8 +44,7 @@ except ModuleNotFoundError: from pkg_resources import parse_version -ZIP_COMPRESSION = zipfile.ZIP_BZIP2 -ZIP_COMPRESSION_LEVEL = 9 + _LOGGER = inkBoard.getLogger(__name__) @@ -124,7 +129,7 @@ def compare_versions(requirement: Union[str,"Version"], compare_version: Union[s comp_str = f"compare_version {c[0]} required_version" else: req_version = requirement - comp_str = f"compare_version >= required_version" + comp_str = "compare_version >= required_version" return eval(comp_str, {}, {"compare_version": compare_version, "required_version": parse_version(req_version)}) diff --git a/inkBoard/arguments.py b/inkBoard/arguments.py index 9918c82..f7948c0 100644 --- a/inkBoard/arguments.py +++ b/inkBoard/arguments.py @@ -34,7 +34,7 @@ def command_logs(args): def command_designer(args): if not DESIGNER_MOD: - print("Running inkBoard designer require the inkBoard designer to be installed") + print("Running inkBoard designer requires the inkBoard designer to be installed") print("Run 'pip install inkBoarddesigner' to install it") return 1 @@ -53,7 +53,7 @@ def command_pack(args): const.COMMAND_VERSION: command_version, const.COMMAND_DESIGNER: command_designer, const.COMMAND_INSTALL: command_install, - const.COMMAND_LOGS: command_logs + const.COMMAND_LOGS: command_logs, } "Action that can/have to be run before creating the CORE object" diff --git a/inkBoard/logging.py b/inkBoard/logging.py index 21b5ac6..2abd6f0 100644 --- a/inkBoard/logging.py +++ b/inkBoard/logging.py @@ -3,7 +3,7 @@ import logging import logging.handlers from pathlib import Path -from typing import Any, Optional, TYPE_CHECKING, Union, TypedDict +from typing import Any, Optional, TYPE_CHECKING, Union, TypedDict, Callable from functools import partial, partialmethod from contextlib import suppress from dataclasses import asdict @@ -119,11 +119,14 @@ class BaseFormatter(logging.Formatter): formatter = logging.Formatter(log_format, log_dateformat, style="$") @classmethod - def format(cls, record): - if not "YAML" in record.__dict__: + def format(cls, record, format_func : Callable = None): + #[ ]: This should be a method. And return super().format + if "YAML" not in record.__dict__: record.__dict__["YAML"] = "" else: record.__dict__["YAML"] = cls._format_yaml(record.__dict__["YAML"]) + if format_func: + return format_func(record) return cls.formatter.format(record) @staticmethod @@ -159,7 +162,7 @@ def __init__(self, fmt = log_format, datefmt = log_dateformat, style = "$", vali super().__init__(fmt, datefmt, style, validate) def format(self, record): - formatted = BaseFormatter.format(record) + formatted = BaseFormatter.format(record, super().format) if record.levelno < DEBUG: format_level = 0 elif record.levelno in ANSI_FORMATS: diff --git a/inkBoard/packaging/__init__.py b/inkBoard/packaging/__init__.py new file mode 100644 index 0000000..508605d --- /dev/null +++ b/inkBoard/packaging/__init__.py @@ -0,0 +1,113 @@ +"Handles inkBoard packages, both creating and installing them." + +import asyncio +from typing import TYPE_CHECKING, Union +from pathlib import Path + +import inkBoard +# from inkBoard.types import * +from inkBoard import constants as bootstrap +from .types import internalinstalltypes + + +if TYPE_CHECKING: + from inkBoard import CORE as CORE + +_LOGGER = inkBoard.getLogger(__name__) +_LOGGER.warning("Dont forget to write tests fro this") + +def confirm_input(msg: str): + answer = input(f"{msg}\n(Y/N): ") + if answer.lower() in {"y","yes"}: + return True + elif answer.lower() in {"n","no"}: + return False + else: + print("Please answer one of Y(es) or N(o) (Not case sensitive)") + return confirm_input(msg) + +def create_config_package(configuration: str, name: str = None, pack_all: bool = False, config: bool = False, platform: bool = False, integrations: bool = False): + """Sets up a core instance and creates a package from it + + Parameters + ---------- + configuration : str + The YAML file to use + name : str, optional + The name of the package, by default None + pack_all : bool, optional + Packages all components (config stuff, platform and integrations), by default False + config : bool, optional + Packages the config folder, by default False + platform : bool, optional + Packages the platform, by default False + integrations : bool, optional + Packages the imported integrations, by default False + + Returns + ------- + int + Return code + """ + core = asyncio.run(bootstrap.setup_core(configuration, bootstrap.loaders.IntegrationLoader)) + return create_core_package(core, name, pack_all, config, platform, integrations) + +def create_core_package(core: "CORE", name: str = None, pack_all: bool = False, config: bool = False, platform: bool = False, integrations: bool = False): + """Creates an inkBoard package from a core instance. + + This bundles all required files and folders from the configuration folder, as well in the required platforms and integrations. + + Parameters + ---------- + core : CORE + The core object constructed from the config + """ + from .package import Packager + + if pack_all: + Packager(core).create_package(name) + else: + pack = [] + if config: pack.append("configuration") + if platform: pack.append("platform") + if integrations: pack.append('integration') + + Packager(core).create_package(name, pack) + return 0 + + +def run_install_command(file: str, name: str, no_input: bool): + ##Add functionality to installer for internal installs (platforms and integrations) + ##Usage: install [platform/integration] [name] + + if file in internalinstalltypes.__args__: + return install_internal(file, name, no_input) + else: + return install_packages(file, no_input) + +def install_internal(install_type: str, name:str, no_input: bool = False): + from .install import InternalInstaller + return InternalInstaller(install_type, name, no_input, confirm_input).install() + +def install_packages(file: Union[str, Path] = None, no_input: bool = False): + + ##https://gist.github.com/oculushut/193a7c2b6002d808a791 + ##Found this gist, that may allow downloading files from github + ##Would make installing integrations and platforms A LOT more user friendly + ##Especially when simply needing to update + ##Better yet, see this code: https://github.com/fbunaren/GitHubFolderDownloader + ##Only relies on requests. That is a library I am a-okay with implementing + from .install import PackageInstaller + if file: + return PackageInstaller(file, skip_confirmations=no_input, confirmation_function=confirm_input).install() + else: + packages = PackageInstaller.gather_inkboard_packages() + + print(f"Found {len(packages)} {'package' if len(packages) == 1 else 'packages'} that can be installed") + for package in packages: + ##Add a confirmation message for each file. + PackageInstaller(package, skip_confirmations=no_input, confirmation_function=confirm_input).install() + return 0 + + + diff --git a/inkBoard/packaging/constants.py b/inkBoard/packaging/constants.py new file mode 100644 index 0000000..30c9ca1 --- /dev/null +++ b/inkBoard/packaging/constants.py @@ -0,0 +1,33 @@ + +from types import MappingProxyType +import zipfile + +from .types import ( + packagetypes, +) + +INKBOARD_PACKAGE_INTERNAL_FOLDER = ".inkBoard" +#Folder name where files from a package are put which are gotten from or destined to the site-packages inkBoard folder. + +PACKAGE_INDEX_URL = "https://github.com/Slalamander/inkBoard-package-index" +"url to the package index" + +INTERNAL_PACKAGE_INDEX_FILE = "package_index.json" + +ZIP_COMPRESSION = zipfile.ZIP_BZIP2 +ZIP_COMPRESSION_LEVEL = 9 + +PACKAGE_ID_FILES : dict[packagetypes,str] = MappingProxyType({ + 'package': 'package.json', + 'integration': 'manifest.json', + 'platform': 'platform.json' +}) + +VERSION_COMPARITORS = ('==', '!=', '>=', '<=', '>', '<') ##The order of this is important! +"Comparison operators allowed for versioning, so they can be evaluated internally" + +DESIGNER_FILES = {"designer", "designer.py"} +#Files in integrations etc. meant for the designer. Currently these are not included in files in the package index + +REQUIREMENTS_FILE = 'requirements.txt' + diff --git a/inkBoard/packaging/download.py b/inkBoard/packaging/download.py new file mode 100644 index 0000000..b863d51 --- /dev/null +++ b/inkBoard/packaging/download.py @@ -0,0 +1,164 @@ +"""Functions to download stuff from the package index +Trying to implement this without using requests +""" +from typing import ( + Union, + Literal, + TYPE_CHECKING +) +import urllib.request +from pathlib import Path +import os +from datetime import datetime as dt +from datetime import timedelta +import json + +from inkBoard import logging + +from .types import ( + PackageIndex +) +from .constants import ( + PACKAGE_INDEX_URL, + INTERNAL_PACKAGE_INDEX_FILE, +) + +if TYPE_CHECKING: + from .version import Version + +_LOGGER = logging.getLogger(__name__) +##For the logger here, maybe use a seperate format? +##Or at least for the info logs + +##See this repo: https://github.com/fbunaren/GitHubFolderDownloader +##And this gist: https://gist.github.com/oculushut/193a7c2b6002d808a791 + +packagebranchtypes = ("main", "dev") + +class Downloader: + + destination_folder : Path + "Folder where downloaded files are put. Usually a temporary folder" + + index : PackageIndex + "inkBoard package index with information on package versions" + + _index_downloaded : bool = False + #Indicate whether the index has been updated for this instance of the downloader + #Means it can be updated i.e. when asking to download a package of which the version is supposedly already up to date + + ##Some stuff to test: + ##Invalid url (i.e. invalid name thing) + ##No internet + ##Mainly just to know what the errors are + def __init__(self, destination_folder : Path): + print(self._index_downloaded) + self.destination_folder = destination_folder + + def get_package_index(self, force_get = False) -> dict: + + ##Internal file is NOT going to be put into the destination folder + internal_file = self.destination_folder / INTERNAL_PACKAGE_INDEX_FILE + + get_index = False + if force_get: + get_index = True + elif self._index_downloaded: + get_index = False + elif internal_file.exists(): + ##Check last changed (i.e. downloaded) time? + last_change = os.path.getmtime(internal_file) + d = dt.now() - dt.fromtimestamp(last_change) + if d.days != 0: + get_index = True + else: + get_index = True + + if get_index: + self._download_package_index() + self._index_downloaded = True + + with open(internal_file) as f: + package_index = PackageIndex(json.load(f)) + + self.index : PackageIndex = package_index + return package_index + + def download_integration_package(self, name : str, package_type : Literal["main","dev"] = "main", version : str = ""): + #"https://github.com/Slalamander/inkBoard-package-index/raw/refs/heads/main/platforms/desktop0.0.2.zip" + if package_type not in ("main","dev"): + raise ValueError(f"branch must be one of {packagebranchtypes}, you passed {package_type}") + + if not hasattr(self,"index"): + self.get_package_index() + + if name not in self.index["integrations"]: + raise KeyError(f"Integration {name} is not known in the package index") + + if version: + ##Handle cases where _dev should be appended to it? + _LOGGER.warning(f"Downloading custom versions is not available (yet). Download of {name} will fail if the version is not the newest one in the
or branch.") + if package_type == "dev" or version.count(".") >= 3: + ##main versions should be 0.2.5 etc. For dev, it may be 0.2.5.dev1 etc. + ##If it's more something went wrong tbf lol + ##Minor versions **should** end up in the main index too. + _LOGGER.debug(f"appending _dev suffix to {name} version {version} for package handling") + fileversion = f"{version}_dev" + raise ValueError("I don't think it should be handled like this for specific versions") + ##Basically: for versions if specified, check if it is the main version -> prefer + ##If dev version -> well you know it's the dev version + ##Otherwise, there will be a dev/main folder in the repo that handles those and which should be set via the package_type honestly? + ##Or should there be a check in the indexer whether a dev version is elligeble? + + pass + else: + fileversion = self.index["integrations"][name][package_type] + if package_type == "dev": + # version = + fileversion = f"{fileversion}_dev" + pass + + raw_url = self._make_raw_file_link() + + #[ ] for platforms, ask to copy files like readme etc. into the current working directory? + return + + @classmethod + def _download_package_index(cls, *, _destination_file : Union[Path, str] = None): + if _destination_file == None: + _destination_file = Path(__file__).parent / INTERNAL_PACKAGE_INDEX_FILE + _LOGGER.info("Getting inkBoard package index from github") + raw_url = cls._make_raw_file_link("index.json") + cls._download_raw_file(raw_url, _destination_file) + _LOGGER.info("Updated inkBoard package index") + + + @staticmethod + def _download_raw_file(raw_file_url : str, destination_file : str): + + filename, headers = urllib.request.urlretrieve(raw_file_url, filename=destination_file) + _LOGGER.debug(f"Successfully downloaded {filename}") + return filename, headers + + @staticmethod + def _make_raw_file_link(file : str, branch = "main", repo_url : str = PACKAGE_INDEX_URL) -> str: + """Returns the link to the raw file on github + + Parameters + ---------- + file : str + The path to the file to download + branch : str, optional + The brand of the repo to get the file from, by default "main" + repo_url : str + url to the repo. Defaults to the inkBoard package index + + Returns + ------- + str + url to the raw version (raw.githubusercontent) of the file + """ + assert repo_url.startswith("https://github.com"), "repo_url must start with `https://github.com`" + repo_url = repo_url.removesuffix("/") + raw_base_url = repo_url.replace("github.com", "raw.githubusercontent.com") + return raw_base_url + "/refs/heads/" + branch + "/" + file diff --git a/inkBoard/packaging/install.py b/inkBoard/packaging/install.py new file mode 100644 index 0000000..b4ef82e --- /dev/null +++ b/inkBoard/packaging/install.py @@ -0,0 +1,819 @@ +"Functions for installing inkBoard packages" + +from typing import ( + TYPE_CHECKING, + Callable, + Union, + Optional, +) +from abc import abstractmethod +from contextlib import suppress +import json +import subprocess +import sys +import zipfile +import os +import tempfile +import shutil +from pathlib import Path + +from inkBoard import logging +from inkBoard.types import ( + platformjson, + manifestjson, + inkboardrequirements, +) +from inkBoard.constants import ( + INKBOARD_FOLDER, + CONFIG_FILE_TYPES, + DESIGNER_INSTALLED, + DESIGNER_FOLDER +) + +from .types import ( + packagetypes, + PackageDict, + NegativeConfirmation, + internalinstalltypes, +) +from .constants import ( + PACKAGE_ID_FILES, + INKBOARD_PACKAGE_INTERNAL_FOLDER, + REQUIREMENTS_FILE +) +from .version import ( + InkboardVersion, + PSSMVersion, + parse_version, + compare_versions, + get_comparitor_string, + ) + +_LOGGER = logging.getLogger(__name__) + +class BaseInstaller: + """Base class for installers + + Call `Installer().install()` to run the installer, or use the pip functions to install packages via pip + """ + + _skip_confirmations: bool + _confirmation_function: Callable[[str, 'BaseInstaller'],bool] + + @property + def skip_confirmations(self) -> bool: + "Whether to ask for confirmation for all actions" + return self._skip_confirmations + + @property + def confirmation_function(self) -> Callable[[str, 'BaseInstaller'],bool]: + "The function used to prompt the user for confirmation" + return self._confirmation_function + + + @abstractmethod + def install(self): + "Runs the installer" + return + + def install_platform_requirements(self, name: str, platform_conf: platformjson) -> bool: + """Installs requirements based on a platformjson dict + + Parameters + ---------- + name : str + Name of the platform. For logging + platform_conf : platformjson + The platform.json dict + + Returns + ------- + bool + Whether the requirements were installed successfully + """ + + platform = name + requirements = platform_conf["requirements"] + if requirements: + res = self.pip_install_packages(*requirements, no_input=self._skip_confirmations) + + if res.returncode != 0: + try: + msg = f"Something went wrong installing the requirements using pip. Continue installation of platform {platform}?" + self.ask_confirm(msg, force_ask=True) + except NegativeConfirmation: + return False + + for opt_req, reqs in platform_conf.get("optional_requirements", {}).items(): + with suppress(NegativeConfirmation): + msg = f"Install requirements for optional features {opt_req}?" + self.ask_confirm(msg) + self.pip_install_packages(*reqs, no_input=self._skip_confirmations) + + return True + + def install_integration_requirements(self, name: str, manifest: manifestjson) -> bool: + """Installs integration requirements based on a manifestjson dict + + Parameters + ---------- + name : str + Name of the integration. For logging + platform_conf : platformjson + The manifest.json dict + + Returns + ------- + bool + Whether the requirements were installed successfully + """ + + integration_version = parse_version(manifest['version']) + integration = name + + _LOGGER.info(f"Installing new Integration {integration}, version {integration_version}") + + requirements = manifest["requirements"] + if requirements: + res = self.pip_install_packages(*requirements, no_input=self._skip_confirmations) + + if res.returncode != 0: + try: + msg = f"Something went wrong installing the requirements using pip. Continue installation of integration {integration}?" + self.ask_confirm(msg, force_ask=True) + except NegativeConfirmation: + return False + + for opt_req, reqs in manifest.get("optional_requirements", {}).items(): + with suppress(NegativeConfirmation): + msg = f"Install requirements for optional features {opt_req}?" + self.ask_confirm(msg) + self.pip_install_packages(*reqs, no_input=self._skip_confirmations) + return True + + def ask_confirm(self, msg: str, force_ask: bool = False): + """Prompts the user to confirm something. + + Calls the confirmation function passed at initialising if skip_confirmations is `False` + + Parameters + ---------- + msg : str + The message to pass to the confirmation function + force_ask : bool + Force the prompt to appear, regardless of the value passed to `skip_confirmations` + + Raises + ------ + NegativeConfirmation + Raised if the confirmation does not evaluate as `True` + """ + if self._skip_confirmations and not force_ask: + return + + if self._confirmation_function: + if not self._confirmation_function(msg, self): + raise NegativeConfirmation + return + + def check_inkboard_requirements(self, ib_requirements: inkboardrequirements, required_for: str) -> bool: + """Checks if inkBoard requirements are met for the current install + + Performs version checks for inkBoard and pssm, and checks for installed platforms and their versions. + + Parameters + ---------- + ib_requirements : inkboardrequirements + Dict with inkBoard specific requirements + required_for : str + What the requirements are required for. Used in log messages. Best practice is to pass it as [type] [name], e.g. 'Platform desktop' + + Returns + ------- + bool + `True` if requirements are met, otherwise `False`. + """ + + ##Check: required inkboard version, pssm version and required integrations/platforms + warn = False + if v := ib_requirements.get("inkboard_version", None): + if not compare_versions(v, InkboardVersion): + warn = True + _LOGGER.warning(f"{required_for} requirment for inkBoard's version not met: {v}") + + if v := ib_requirements.get("pssm_version", None): ##I think this should generally be met by having the inkBoard requirement met tho? + if not compare_versions(v, PSSMVersion): + warn = True + _LOGGER.warning(f"{required_for} requirment for PSSM's version not met: {v}") + + for platform in ib_requirements.get('platforms', []): + req_vers = None + if c := get_comparitor_string(platform): + platform, req_vers = platform.split(c) + + if not (INKBOARD_FOLDER / "platforms" / platform).exists(): + warn = True + _LOGGER.warning(f"Platform {platform} required for {required_for} is not installed") + ##Should maybe check this in regards with package installing? i.e. if these are otherwise present in the package + ##But will come later. + elif req_vers: + with open(INKBOARD_FOLDER / "platforms" / platform / PACKAGE_ID_FILES["platform"]) as f: + platform_conf: platformjson = json.load(f) + cur_version = platform_conf["version"] + + if not compare_versions(c + req_vers, cur_version): + warn = True + _LOGGER.warning(f"Platform {platform} does not meet the version requirement: {c + req_vers}") + + ##And do the same for integrations. + for integration in ib_requirements.get('integrations', []): + req_vers = None + if c := get_comparitor_string(integration): + integration, req_vers = integration.split(c) + + if not (INKBOARD_FOLDER / "integrations" / integration).exists(): + warn = True + _LOGGER.warning(f"Integration {integration} required for {required_for} is not installed") + ##Should maybe check this in regards with package installing? i.e. if these are otherwise present in the package + ##But will come later. + elif req_vers: + with open(INKBOARD_FOLDER / "integrations" / integration / PACKAGE_ID_FILES["integration"]) as f: + integration_conf: manifestjson = json.load(f) + cur_version = integration_conf["version"] + + if not compare_versions(c + req_vers, cur_version): + warn = True + _LOGGER.warning(f"Integration {integration} does not meet the version requirement: {c + req_vers}") + + return not warn + + @staticmethod + def pip_install_packages(*packages: str, no_input: bool = False) -> subprocess.CompletedProcess: + """Calls the pip command to install the provided packages + + Parameters + ---------- + packages : str + The packages to install (as would be passed to pip as arguments) + no_input: bool + Disables prompts from pip + + Returns + ------- + subprocess.CompletedProcess + The result of the subprocess.run function + """ + + if not packages: + return + + if no_input: + args = [sys.executable, '-m', 'pip', '--no-input', 'install', *packages] + else: + args = [sys.executable, '-m', 'pip', 'install', *packages] + + res = subprocess.run(args) + return res + + @staticmethod + def pip_install_requirements_file(file: Union[str,Path], *, no_input: bool = False) -> subprocess.CompletedProcess: + """Calls the pip command to install the provided .txt file with requirements + + Parameters + ---------- + file : Union[str,Path] + The text file holding the requirements + no_input: bool + Disables prompts from pip + + Returns + ------- + subprocess.CompletedProcess + The result of the subprocess.run function + """ + + if isinstance(file,Path): + file = str(file.resolve()) + + if no_input: + args = [sys.executable, '-m', 'pip', '--no-input', 'install', '-r', file] + else: + args = [sys.executable, '-m', 'pip', 'install', '-r', file] + + + res = subprocess.run(args) + return res + + ##Options to install: + # - Package + # - Platform + # - Integration + # - requirements; internal and external -> internal eh, should be taken care of when actually installing it. + +class PackageInstaller(BaseInstaller): + """Installs an inkBoard compatible .zip file, or requirements files in a config directory. + + Call `PackageInstaller().install()` to run the installer. + There are a few classmethods and staticmethods too that can be called without instantiating. + + Parameters + ---------- + file : Union[Path,str] + The .zip file to install + skip_confirmations : bool, optional + Skips most confirmation messages during installation, except those deemed vital, by default False + confirmation_function : Callable[[str, Installer],bool], optional + Function to call when asking for confirmation, gets passed the question to confirm and the Installer instance., by default None + """ + + def __init__(self, file: Union[Path,str], skip_confirmations: bool = False, confirmation_function: Callable[[str, 'BaseInstaller'],bool] = None): + self._file = Path(file) + assert self._file.exists(), f"{file} does not exist" + self._confirmation_function = confirmation_function + self._skip_confirmations = skip_confirmations + + if self._file.suffix in CONFIG_FILE_TYPES: + self._package_type = "configuration" + else: + self._package_type: packagetypes = self.identify_zip_file(self._file) + return + + def install(self): + """Runs the appropriate installer for the package type. + """ + + if self._package_type == "integration": + self.install_integration() + elif self._package_type == "platform": + self.install_platform() + elif self._package_type == "package": + self.install_package() + elif self._package_type == "configuration": + self.install_config_requirements(self._file, self._skip_confirmations, self._confirmation_function) + + def install_package(self) -> Optional[packagetypes]: + """Installs a package type .zip file file + """ + + file = self._file + + if self._package_type != 'package': + raise TypeError(f"{file} is not a package type .zip file") + + try: + self.ask_confirm(f"Install package {file}?") + except NegativeConfirmation: + _LOGGER.info(f"Not installing package {file}") + return + + with zipfile.ZipFile(file) as zip_file: + self.__zip_file = zip_file + zip_path = zipfile.Path(zip_file) + # with zip_file.open(packageidfiles["package"]) as f: + ##This section is used to determine compatibility of the package and the installed modules + f = zip_file.open(PACKAGE_ID_FILES["package"]) + package_info: PackageDict = json.load(f) + + vers_msg = "" + if (v := parse_version(package_info["versions"]["inkBoard"])) >= InkboardVersion: + vers_msg = vers_msg + f"Package was made with a newer version of inkBoard ({v}). Installed is {InkboardVersion}." + + if (v := parse_version(package_info["versions"]["PythonScreenStackManager"])) >= PSSMVersion: + vers_msg = vers_msg + f"Package was made on with a newer version of PSSM ({v}). Installed is {PSSMVersion}." + + if vers_msg: + print(vers_msg) + try: + self.ask_confirm(f"Version mismatch, continue installing {self._package_type}?") + except NegativeConfirmation: + return + + ##Check if platform is installed or present in the package. + package_platform = package_info["platform"] + + if ((INKBOARD_FOLDER / "platforms" / package_platform).exists() or + (zip_path / INKBOARD_PACKAGE_INTERNAL_FOLDER / "platforms" / package_platform).exists()): + pass + else: + msg = f"Package was made for platform {package_platform}, but it is not installed or present in the package. Continue installing?" + try: + self.ask_confirm(msg) + except NegativeConfirmation: + return + + if (zip_path / INKBOARD_PACKAGE_INTERNAL_FOLDER / "platforms").exists(): + _LOGGER.info("Installing platforms") + for platform_folder in (zip_path / INKBOARD_PACKAGE_INTERNAL_FOLDER / "platforms").iterdir(): + + with suppress(NegativeConfirmation): + self.ask_confirm(f"Install platform {platform_folder.name}?") + try: + _LOGGER.info(f"Installing platform {platform_folder.name}") + self._install_platform_zipinfo(zip_file.getinfo(platform_folder.at)) + except NegativeConfirmation: + pass + except Exception as exce: + _LOGGER.error(f"Could not install platform {platform_folder.name}", exc_info=exce) + _LOGGER.info("Platforms installed") + + + if (zip_path / INKBOARD_PACKAGE_INTERNAL_FOLDER / "integrations").exists(): + _LOGGER.info("Installing integrations") + for integration_folder in (zip_path / INKBOARD_PACKAGE_INTERNAL_FOLDER / "integrations").iterdir(): + + with suppress(NegativeConfirmation): + self.ask_confirm(f"Install integration {integration_folder.name}?") + try: + _LOGGER.info(f"Installing integration {integration_folder.name}") + self._install_integration_zipinfo(zip_file.getinfo(integration_folder.at)) + except Exception as exce: + _LOGGER.error(f"Could not install integration {integration_folder.name}", exc_info=exce) + _LOGGER.info("Integrations installed") + + if (zip_path / "configuration").exists(): + _LOGGER.info(f"Extracting configuration folder to current working directory {Path.cwd()}") + ##First extract, then find requirements.txt file + + self.extract_zip_folder(zip_file.getinfo((zip_path / "configuration").at), + allow_overwrite=True, just_contents=True) + + self.install_config_requirements(Path.cwd()) + + _LOGGER.info("Package succesfully installed") + return + + def install_platform(self): + """Installs a platform type .zip file file + """ + file = self._file + + if self._package_type != 'platform': + raise TypeError(f"{file} is not a platform type .zip file") + + with zipfile.ZipFile(file) as zip_file: + self.__zip_file = zip_file + zip_path = zipfile.Path(zip_file) + p = list(zip_path.iterdir())[0] + self._install_platform_zipinfo(zip_file.getinfo(p.at)) + + return + + def install_integration(self): + """Installs an integration type .zip file file + """ + + file = self._file + + if self._package_type != 'integration': + raise TypeError(f"{file} is not an integration type .zip file") + + with zipfile.ZipFile(file) as zip_file: + self.__zip_file = zip_file + zip_path = zipfile.Path(zip_file) + p = list(zip_path.iterdir())[0] + self._install_integration_zipinfo(zip_file.getinfo(p.at)) + + return + + def install_config_requirements(self, config_file: Union[str,Path]): + """Installs requirements for the passed config_file + + If config_file is a .yaml file it will use the folder the file is in. If is it a folder, that folder be used as a base. + The function looks for requirements.txt files in the 'config folder' itself, in 'config folder/files' (but only the top folder), and recursively in 'config folder/custom' (i.e. it goes through all files in all folders within there and installs every requirements.txt it finds) + Afterwards, it will look in 'config folder/custom/integrations' and install all requirements for all integrations, as well as prompt for optional requirements if a function is supplied. + + Parameters + ---------- + config_file : Union[str,Path] + The yaml file from which to get the base folder, or the base folder itself + skip_confirmations : bool, optional + Instructs pip to not prompt for confirmations when installing, by default False + confirmation_function : Callable[[str],bool], optional + Function to call when optional requirements can be installed, by default `confirm_input` (command line prompt). If a boolean `False` is returned, or a `NegativeConfirmation` error is raised, the optional requirements are not installed. + """ + + skip_confirmations = self._skip_confirmations + + if isinstance(config_file,str): + config_file = Path(config_file) + + if config_file.is_file(): + assert config_file.suffix in CONFIG_FILE_TYPES, "Config file must be a yaml file" + path = config_file.parent + else: + path = config_file + + if (path / "custom").exists(): + if (path / REQUIREMENTS_FILE).exists(): + self.pip_install_requirements_file(path / REQUIREMENTS_FILE, skip_confirmations) + + if (path / "files" / REQUIREMENTS_FILE).exists(): + self.pip_install_packages(path / "files" / REQUIREMENTS_FILE, skip_confirmations) + + folder = path / "custom" + for foldername, subfolders, filenames in os.walk(folder): + if REQUIREMENTS_FILE in filenames: + file_path = os.path.join(foldername, REQUIREMENTS_FILE) + self.pip_install_requirements_file(file_path, skip_confirmations) + + if (folder / "integrations").exists(): + for integration_folder in (folder / "integrations").iterdir(): + with open(integration_folder / PACKAGE_ID_FILES["integration"]) as f: + integration_conf: manifestjson = json.load(f) + + if reqs := integration_conf.get("requirements", []): + _LOGGER.info(f"Installing requirements for custom integration {integration_folder.name}") + res = self.pip_install_packages(*reqs, no_input=skip_confirmations) + + if res.returncode != 0: + _LOGGER.error(f"Something went wrong installing requirements for custom integration {integration_folder.name}") + continue + + for opt_req, reqs in integration_conf.get("optional_requirements", {}).items(): + with suppress(NegativeConfirmation): + msg = f"Install requirements for optional features {opt_req} for custom integration {integration_folder.name}?" + + if self._confirmation_function: + if not self._confirmation_function(msg, self): + continue + self.pip_install_packages(*reqs, no_input=skip_confirmations) + + def _install_platform_zipinfo(self, platform_info: zipfile.ZipInfo): + + assert platform_info.is_dir(),"Platforms must be a directory" + + platform_zippath = zipfile.Path(self.__zip_file, platform_info.filename) + platform = platform_zippath.name + + f = self.__zip_file.open(f"{platform_info.filename}{PACKAGE_ID_FILES['platform']}") + platform_conf: platformjson = json.load(f) + platform_version = parse_version(platform_conf['version']) + + install = True + ib_requirements = platform_conf["inkboard_requirements"] + + if not self.check_inkboard_requirements(ib_requirements, f"Platform {platform}"): + msg = f"inkBoard requirements for platform {platform} are not met (see logs). Continue installing?" + self.ask_confirm(msg) + + if (INKBOARD_FOLDER / "platforms" / platform).exists(): + + with open(INKBOARD_FOLDER / "platforms" / platform / PACKAGE_ID_FILES["platform"]) as f: + cur_conf: platformjson = json.load(f) + cur_version = parse_version(cur_conf['version']) + + if cur_version > platform_version: + msg = f"Version {cur_version} of platform {platform} is currently installed. Do you want to install earlier version {platform_version}?" + self.ask_confirm(msg) + + elif platform_version > cur_version: + _LOGGER.info(f"Updating platform {platform} from version {cur_version} to {platform_version}.") + else: + msg = f"Version {platform_version} of platform {platform} is already installed. Do you want to overwrite it?" + self.ask_confirm(msg) + + if not install: + _LOGGER.info(f"Not installing platform {platform} {platform_version}") + return + + _LOGGER.info(f"Installing new platform {platform}, version {platform_version}") + if self.install_platform_requirements(platform,platform_conf): + self.extract_zip_folder(platform_info, path = INKBOARD_FOLDER / "platforms", allow_overwrite=True) + _LOGGER.info("Extracted platform file") + return + + def _install_integration_zipinfo(self, integration_info: zipfile.ZipInfo): + + assert integration_info.is_dir(),"Integrations must be a directory" + + integration_zippath = zipfile.Path(self.__zip_file, integration_info.filename) + integration = integration_zippath.name + + manifestpath = integration_zippath / PACKAGE_ID_FILES['integration'] + f = manifestpath.open() + integration_conf: manifestjson = json.load(f) + integration_version = parse_version(integration_conf['version']) + + install = True + ib_requirements = integration_conf.get("inkboard_requirements",{}) + + if ib_requirements and not self.check_inkboard_requirements(ib_requirements, f"Integration {integration}"): + msg = f"inkBoard requirements for integration {integration} are not met (see logs). Continue installing?" + self.ask_confirm(msg) + + if (INKBOARD_FOLDER / "integrations" / integration).exists(): + + with open(INKBOARD_FOLDER / "integrations" / integration / PACKAGE_ID_FILES["integration"]) as f: + cur_conf: manifestjson = json.load(f) + cur_version = parse_version(cur_conf['version']) + + if cur_version > integration_version: + msg = f"Version {cur_version} of Integration {integration} is currently installed. Do you want to install earlier version {integration_version}?" + self.ask_confirm(msg) + + elif integration_version > cur_version: + _LOGGER.info(f"Updating Integration {integration} from version {cur_version} to {integration_version}.") + else: + msg = f"Version {integration_version} of Integration {integration} is already installed. Do you want to overwrite it?" + self.ask_confirm(msg) + + if not install: + _LOGGER.info(f"Not installing Integration {integration} {integration_version}") + return + + if self.install_integration_requirements(integration, integration_conf): + self.extract_zip_folder(integration_info, path = INKBOARD_FOLDER / "integrations", allow_overwrite=True) + _LOGGER.info("Extracted integration files") + + def extract_zip_folder(self, member: Union[str,zipfile.ZipInfo], path: Union[str,Path,None] = None, pwd: str = None, just_contents: bool = False, allow_overwrite: bool = False): + """Extracts a folder and all it's contents from a ZipFile object to path. + + Parameters + ---------- + member : zipfile.ZipInfo + The name or ZipInfo object of the folder to extract + path : _type_, optional + The path to extract the folder to, by default None (which extracts to the current working directory) + just_contents : bool + Extract the contents of the folder directly to path, instead of extracting the folder itself, defaults to `False` + pwd : str, optional + Optional password for the archive, by default None + allow_overwrite : bool + Allows overwriting existing files or folders + """ + + if isinstance(member, str): + member = self.__zip_file.getinfo(member) + + assert member.is_dir(),"Member must be a directory" + + ##Gotta put it all in a temporary directory to isolate the folder correctly + with tempfile.TemporaryDirectory() as tempdir: + self.__zip_file.extract(member, tempdir, pwd) + for file in self.__zip_file.namelist(): + if file.startswith(member.orig_filename) and file != member.orig_filename: + self.__zip_file.extract(file, tempdir, pwd) + _LOGGER.verbose(f"Extracted folder {member.orig_filename} to temporary directory") + + if path == None: + path = Path.cwd() + + if just_contents: + src = Path(tempdir) / Path(member.orig_filename) + else: + src = Path(tempdir) / Path(member.orig_filename).parent + + shutil.copytree( + src = src, + dst = path, + dirs_exist_ok = allow_overwrite + ) + _LOGGER.debug("Copied from tempdir") + ##What happens here with nested stuff? I.e. internal folders -> check with package extraction + + return + + @classmethod + def gather_inkboard_packages(cls) -> dict[Path, packagetypes]: + """Gathers all inkBoard viable packages availables in the current working directory + + Returns + ------- + dict[Path, packagetypes] + Dict with all the path objects of all found packages and their package type + """ + + _LOGGER.info(f"Gathering inkBoard zip packages in {Path.cwd()}") + + packs = {} + for file in Path.cwd().glob('*.zip'): + if file.suffix != ".zip": + continue + + if p := cls.identify_zip_file(file): + packs[file] = p + + _LOGGER.info(f"Found {len(packs)} inkBoard installable zip packages.") + return packs + + @classmethod + def identify_zip_file(cls, file: Union[str, Path, zipfile.ZipFile]) -> Optional[packagetypes]: + """Identifies the type of inkBoard package the zip file is + + Parameters + ---------- + file : Union[str, Path] + The file to identify. Must be a .zip file. + + Returns + ------- + Optional[packagetypes] + The type of package this file is (package, integration or platform), or None if it could not be identified. + + Raises + ------ + TypeError + Raised if the provided file is not a zipfile + """ + + + if not isinstance(file, zipfile.ZipFile): + zip_file = cls._path_to_zipfile(file) + zip_file.close() + else: + zip_file = file + + p = zipfile.Path(zip_file) + root_files = [f for f in p.iterdir()] + + if len(root_files) == 1 and root_files[0].is_dir(): + ##Look in the single folder and whether it contains a manifest or platform json + if (root_files[0] / PACKAGE_ID_FILES["integration"]).exists(): + return 'integration' + elif (root_files[0] / PACKAGE_ID_FILES["platform"]).exists(): + return 'platform' + elif (p / PACKAGE_ID_FILES["package"]).exists() and (len(root_files) in {2,3}): ##2 or 3: at least contains package.json, and has .inkBoard and/or configuration folder + return 'package' + + return + + @staticmethod + def _path_to_zipfile(file: Union[str, Path]) -> zipfile.ZipFile: + """Converts a path string or Path object to a `zipfile.ZipFile` object + + Parameters + ---------- + file : Union[str, Path] + The file to open + + Returns + ------- + zipfile.ZipFile + The corresponding ZipFile object + + Raises + ------ + TypeError + Raised if the file is not a .zip file + """ + + if isinstance(file, str): + file = Path(file) + + if file.suffix != '.zip': + raise TypeError("File must be a .zip file") + else: + return zipfile.ZipFile(file, 'r') + +class InternalInstaller(BaseInstaller): + "Handles installing requirements of already installed platforms and integrations." + def __init__(self, install_type: internalinstalltypes, name: str, skip_confirmations = False, confirmation_function = None): + ##May remove the subclassing, but just reuse the usable functions (i.e. seperate out a few funcs.) + ##Also, use the constant designer mod in case something is not found internally. + ##Do give a warning for platforms though, or integrations without a designer module. + if install_type == "integration": + file = Path("integrations") / name + elif install_type == "platform": + file = Path("platforms") / name + + full_path = INKBOARD_FOLDER / file + if not DESIGNER_INSTALLED: + assert full_path.exists(), f"{install_type} {name} is not installed or does not exist" + else: + if not full_path.exists(): + assert (DESIGNER_FOLDER / file).exists(), f"{install_type} {name} is not installed or does not exist" + full_path = DESIGNER_FOLDER / file + + self._name = full_path.name + self._full_path = full_path + self._confirmation_function = confirmation_function + self._skip_confirmations = skip_confirmations + self._install_type = install_type + return + + def install(self): + if self._install_type == "integration": + return self.install_integration() + elif self._install_type == "platform": + return self.install_platform() + + def install_platform(self): + + with open(self._full_path / PACKAGE_ID_FILES["platform"]) as f: + conf: platformjson = json.load(f) + + with suppress(NegativeConfirmation): + msg = f"Install platform {self._name}?" + self.ask_confirm(msg) + return self.install_platform_requirements(self._name, conf) + return 1 + + def install_integration(self): + with open(self._full_path / PACKAGE_ID_FILES["integration"]) as f: + conf: platformjson = json.load(f) + + with suppress(NegativeConfirmation): + msg = f"Install integration {self._name}?" + self.ask_confirm(msg) + return self.install_integration_requirements(self._name, conf) + return 1 diff --git a/inkBoard/packaging/package.py b/inkBoard/packaging/package.py new file mode 100644 index 0000000..adff196 --- /dev/null +++ b/inkBoard/packaging/package.py @@ -0,0 +1,264 @@ +from typing import ( + TYPE_CHECKING, + Union, + Literal, + Callable, +) +from pathlib import Path +from datetime import datetime as dt +import tempfile +import json +import zipfile +import os +from functools import partial +import shutil +import inspect + + +from inkBoard import logging +from inkBoard.constants import CONFIG_FILE_TYPES + +from .constants import ( + ZIP_COMPRESSION, + ZIP_COMPRESSION_LEVEL, + DESIGNER_FILES, + INKBOARD_PACKAGE_INTERNAL_FOLDER, +) +from .types import ( + PackageDict +) + +if TYPE_CHECKING: + from inkBoard import CORE + +_LOGGER = logging.getLogger(__name__) + +class Packager: + """Takes care of creating inkBoard packages from configs + """ + + def __init__(self, core: "CORE", folder: Union[str,Path] = None, progress_func: Callable[[str,str, float],None] = None): + self.CORE = core + self.config = core.config + if folder: + if isinstance(folder,str): folder = Path(folder) + assert folder.is_dir(), "Folder must be a directory" + self.base_folder = folder + else: + self.base_folder = core.config.baseFolder + self._copied_yamls = set() + self.__progress_func = progress_func + + def report_progress(self, stage: str, message: str, progress: float): + "Reports progress to the progress function, if any" + if self.__progress_func: + self.__progress_func(stage, message, progress) + else: + _LOGGER.info(message) + + def create_package(self, package_name: str = None, pack: list[Literal['configuration', 'platform', 'integration']] = ['configuration', 'platform', 'integration']): + + self.report_progress("Start", f"Creating a package for {self.CORE.config.file}", 0) + + self.report_progress("Gathering", "Creating temporary directory", 5) + + with tempfile.TemporaryDirectory(dir=self.base_folder) as tempdir: + + if 'configuration' in pack: + self.report_progress("Configuration", "Copying configuration directory", 10) + self.copy_config_files(tempdir) + + if 'platform' in pack: + self.report_progress("Platform", "Copying platform directory", 30) + self.copy_platform_folder(tempdir) + + if 'integration' in pack: + self.report_progress("Integrations", "Copying included integrations", 50) + self.copy_integrations(tempdir) + + self.report_progress("Package Info", "Creating Package info file", 70) + + package_info = self.create_package_dict() + with open(Path(tempdir) / "package.json", 'w') as f: + json.dump(package_info, f, indent=4) + + if not package_name: + package_name = f'inkBoard_package_{package_info["platform"]}_{self.CORE.config.filePath.stem}' + + self.report_progress("Zip File", "Creating Package zipfile", 75) + _LOGGER.info("Creating package zip file") + + zipname = self.base_folder / f'{package_name}.zip' + with zipfile.ZipFile(zipname, 'w', ZIP_COMPRESSION, compresslevel=ZIP_COMPRESSION_LEVEL) as zip_file: + for foldername, subfolders, filenames in os.walk(tempdir): + _LOGGER.verbose(f"Zipping contents of folder {foldername}") + for filename in filenames: + file_path = os.path.join(foldername, filename) + zip_file.write(file_path, os.path.relpath(file_path, tempdir)) + for dir in subfolders: + dir_path = os.path.join(foldername, dir) + zip_file.write(dir_path, os.path.relpath(dir_path, tempdir)) + + self.report_progress("Done", f"Package created: {zipname}", 100) + _LOGGER.info(f"Package created: {zipname}") + + return + + def copy_config_files(self, tempdir): + "Copies all files and folders from the config directory in to the temporary folder" + + _LOGGER.info(f"Copying files from config folder {self.base_folder}") + config_dir = Path(tempdir) / "configuration" + config_folders_copy = { + "icon", "picture", "font", "custom", "file" + } + + + for folder_attr in config_folders_copy: + ignore_func = partial(self.ignore_files, self.config.folders.custom_folder / "integrations", + ignore_in_baseparent_folder = DESIGNER_FILES) + path: Path = getattr(self.config.folders, f"{folder_attr}_folder") + if not path.exists(): + continue + + _LOGGER.info(f"Copying config folder {path.name}") + shutil.copytree( + src= path, + dst= config_dir / path.name, + ignore=ignore_func + ) + + for yamlfile in self.config.included_yamls: + if Path(yamlfile) in self._copied_yamls: + _LOGGER.debug(f"Yaml file {yamlfile} was already copied.") + continue + + _LOGGER.debug(f"Copying yaml file {yamlfile}") + shutil.copy2( + src=yamlfile, + dst=config_dir + ) + + _LOGGER.info("Succesfully copied contents of config folder.") + return + + def copy_platform_folder(self, tempdir): + + tempdir = Path(tempdir) + + if self.CORE.DESIGNER_RUN: + platform = self.CORE.device.emulated_platform + platform_folder = self.CORE.device.emulated_platform_folder + else: + platform = self.CORE.device.platform + platform_folder = Path(inspect.getfile(self.CORE.device.__class__)).parent + + _LOGGER.info(f"Copying platform {platform} from {platform_folder} to package") + + manual_files = {"readme.md", "install.md", "installation.md", "package_files"} + manual_dir = (tempdir / "configuration") if (tempdir / "configuration").exists() else tempdir + + for file in platform_folder.iterdir(): + if file.name.lower() not in manual_files: + continue + + _LOGGER.debug(f"Copying platform manual file {file}") + if file.is_dir(): + shutil.copytree( + src = file, + dst = manual_dir / "files", + dirs_exist_ok=True + ) + else: + manual_files.add(file.name) + shutil.copy2( + src = file, + dst = manual_dir + ) + + ignore_func = partial(self.ignore_files, platform_folder.parent, ignore_in_baseparent_folder = manual_files | DESIGNER_FILES ) + _LOGGER.debug("Copying platform folder") + shutil.copytree( + src = platform_folder, + dst = tempdir / INKBOARD_PACKAGE_INTERNAL_FOLDER / "platforms" / platform_folder.name, + ignore = ignore_func + ) + + _LOGGER.info("Succesfully copied platform folder") + return + + def copy_integrations(self, tempdir): + + tempdir = Path(tempdir) + + ##Filter out integrations from the custom folder + all_integrations : dict[str,Path] = self.CORE.integrationLoader.imported_integrations + ##E + + _LOGGER.info("Copying all non custom integrations to package") + for integration, location in all_integrations.items(): + if location.is_relative_to(self.config.folders.custom_folder): + ##Skip integrations here. Those were already copied during the config folder phase + continue + _LOGGER.debug(f"Copying integration {integration}") + ignore_func = partial(self.ignore_files, location.parent, ignore_in_baseparent_folder=DESIGNER_FILES) + shutil.copytree( + src= location, + dst= tempdir / INKBOARD_PACKAGE_INTERNAL_FOLDER / "integrations" / location.name, + ignore=ignore_func + ) + + _LOGGER.info("Succesfully copied integrations") + return + + def ignore_files(self, parentbase_folder: Path, src, names, ignore_in_baseparent_folder: set = {}): + """Returns a list with files to not copy for `shutil.copytree` + + Parameters + ---------- + parentbase_folder : Path + The base folder being copied from + src : str + source path, passed by `copytree` + names : list[str] + list with file and folder names, passed by `copytree` + ignore_in_baseparent_folder : set, optional + Set with filenames to ignore (i.e. not copy), _Only if_ the parent folder of `src` is `base_ignore_folder`, by default {} + + Returns + ------- + _type_ + _description_ + """ + + ignore_set = {"__pycache__"} + if Path(src).parent == parentbase_folder: + ignore_set.update(ignore_in_baseparent_folder) + + for name in filter(lambda x: x.endswith(CONFIG_FILE_TYPES), names): + self._copied_yamls.add(Path(src) / name) + + return ignore_set + + def create_package_dict(self) -> PackageDict: + import inkBoard + import PythonScreenStackManager as PSSM + + package_dict = {"created": dt.now().isoformat()} + + package_dict["versions"] = {"inkBoard": inkBoard.__version__, + "PythonScreenStackManager": PSSM.__version__} + + + if self.CORE.DESIGNER_RUN: + import inkBoarddesigner + + package_dict["created_with"] = "inkBoarddesigner" + package_dict["versions"]["inkBoarddesigner"] = inkBoarddesigner.__version__ + package_dict["platform"] = self.CORE.device.emulated_platform + else: + package_dict["created_with"] = "inkBoard" + package_dict["versions"]["inkBoarddesigner"] = None + package_dict["platform"] = self.CORE.device.platform + + return PackageDict(**package_dict) diff --git a/inkBoard/packaging/types.py b/inkBoard/packaging/types.py new file mode 100644 index 0000000..dd6c9c7 --- /dev/null +++ b/inkBoard/packaging/types.py @@ -0,0 +1,46 @@ +from typing import Literal, TypedDict + +internalinstalltypes = Literal["platform", "integration"] +packagetypes = Literal['package', 'integration', 'platform'] + +class PackageDict(TypedDict): + + created: str + "The date and time the package was created, in isoformat" + + created_with: Literal["inkBoard", "inkBoarddesigner"] + "Whether this package was created via inkBoard itself, or via the designer" + + versions: dict[Literal["inkBoard", "PythonScreenStackManager", "inkBoarddesigner"],str] + "The versions of the core packages installed when creating it. Designer version is None if not installed" + + platform: str + "The platform the package was created for" + +class PackageIndex(TypedDict): + "Structure of the index.json file in the package index" + + inkBoard : str + "inkBoard version the index was made on" + + inkBoarddesigner : str + "Version of inkBoard designer the index was made on" + + PythonScreenStackManager : str + "Version of PSSM the index was made on" + + timestamp : str + "ISO format timestamp of when the index file was run" + + platforms : dict[str,dict[Literal["main","dev"],str]] + "List of platforms and their versions in the main and dev branch" + + integrations : dict[str,dict[Literal["main","dev"],str]] + "List of integrations and their versions in the main and dev branch" + +comparisonstrings = Literal[ + '==', '!=', '>=', '<=', '>', '<' +] + +class NegativeConfirmation(UserWarning): + "Raised by ask confirm if the confirmation was negative" \ No newline at end of file diff --git a/inkBoard/packaging/version.py b/inkBoard/packaging/version.py new file mode 100644 index 0000000..5392101 --- /dev/null +++ b/inkBoard/packaging/version.py @@ -0,0 +1,91 @@ +"Helps with comparing versions" + +from typing import TYPE_CHECKING, Union, Literal + +import inkBoard +import PythonScreenStackManager as PSSM + +from .constants import ( + VERSION_COMPARITORS, +) + +from .types import ( + comparisonstrings +) + +try: + from packaging.version import parse as parse_version +except ModuleNotFoundError: + from pkg_resources import parse_version + +if TYPE_CHECKING: + from packaging.version import Version + +InkboardVersion = parse_version(inkBoard.__version__) +PSSMVersion = parse_version(PSSM.__version__) + +def get_comparitor_string(input_str: str) -> Literal[comparisonstrings]: + "Returns the comparitor (==, >= etc.) in a string, or None if there is None." + if c := (x for x in VERSION_COMPARITORS if x in input_str): + return c[0] + return + +def compare_versions(requirement: Union[str,"Version"], compare_version: Union[str,"Version"]) -> bool: + """Does simple version comparisons. + + For requirements, accepts both a general requirement string (i.e. '1.0.0'), or a comparison string (i.e. package < 1.0.0)) + + Parameters + ---------- + requirement : Union[str,"Version"] + The requirement to test + compare_version : Union[str,Version] + The version to compare the requirement to + + Returns + ------- + bool + True if the requirement is satisfied, false if not + """ + + if isinstance(compare_version,str): + compare_version = parse_version(compare_version) + + if not isinstance(requirement, str): + ##To be sure that the pkg_resources Version is also fine + return compare_version >= requirement + + if c := [x for x in VERSION_COMPARITORS if x in requirement]: + req_version = requirement.split(c[0])[-1] ##With how the comparitors are set up, 0 should always be the correct one + comp_str = f"compare_version {c[0]} required_version" + else: + req_version = requirement + comp_str = "compare_version >= required_version" + + return eval(comp_str, {}, {"compare_version": compare_version, "required_version": parse_version(req_version)}) + +def write_version_filename(package_name : str, version : Union[str,"Version"], suffix : str = ".zip") -> str: + """Creates the appropriate filename for an inkBoard index package + + The returned string will have the form of "{package_name}-{version}{suffix}". + If suffix evaluates to a boolean `True`, it will be included, otherwise not. + + Parameters + ---------- + package_name : str + The name of the package + version : str, Version + The version of the package + suffix : str, optional + Suffix for after the strin, by default ".zip" + + Returns + ------- + str + The formatted package name + """ + + if suffix: + return f"{package_name}-{version}{suffix}" + else: + return f"{package_name}-{version}" \ No newline at end of file