From 6172e91068120f615691313c35478f92a770174b Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 13:55:29 +0200 Subject: [PATCH 01/18] =?UTF-8?q?refactor(packaging):=20=F0=9F=8E=A8=20mov?= =?UTF-8?q?e=20packaging=20to=20a=20seperate=20folder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Will use this to split up the functionality better. Keep an eye on if this changes the indexer --- inkBoard/{packaging.py => packaging/__init__.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename inkBoard/{packaging.py => packaging/__init__.py} (100%) diff --git a/inkBoard/packaging.py b/inkBoard/packaging/__init__.py similarity index 100% rename from inkBoard/packaging.py rename to inkBoard/packaging/__init__.py From ac7296cd69efa19358d2292a1713e68f2cda0b12 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:01:55 +0200 Subject: [PATCH 02/18] port constants and types --- inkBoard/packaging/constants.py | 31 ++++++++++++++++++++++++ inkBoard/packaging/types.py | 43 +++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+) create mode 100644 inkBoard/packaging/constants.py create mode 100644 inkBoard/packaging/types.py diff --git a/inkBoard/packaging/constants.py b/inkBoard/packaging/constants.py new file mode 100644 index 0000000..0f6befe --- /dev/null +++ b/inkBoard/packaging/constants.py @@ -0,0 +1,31 @@ + +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" + +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/types.py b/inkBoard/packaging/types.py new file mode 100644 index 0000000..3515cc5 --- /dev/null +++ b/inkBoard/packaging/types.py @@ -0,0 +1,43 @@ +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[ + '==', '!=', '>=', '<=', '>', '<' +] \ No newline at end of file From b8d1698723aea860cbcd58cb3c07bd2020918a8a Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:02:07 +0200 Subject: [PATCH 03/18] port functions related to versioning --- inkBoard/packaging/version.py | 65 +++++++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 inkBoard/packaging/version.py diff --git a/inkBoard/packaging/version.py b/inkBoard/packaging/version.py new file mode 100644 index 0000000..a8dc52d --- /dev/null +++ b/inkBoard/packaging/version.py @@ -0,0 +1,65 @@ +"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)}) From 8ab4f80834fda73f9299bab8d311ee60f6f263cd Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:02:24 +0200 Subject: [PATCH 04/18] save original packaging file --- inkBoard/_packaging.py | 1220 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1220 insertions(+) create mode 100644 inkBoard/_packaging.py diff --git a/inkBoard/_packaging.py b/inkBoard/_packaging.py new file mode 100644 index 0000000..ac3feec --- /dev/null +++ b/inkBoard/_packaging.py @@ -0,0 +1,1220 @@ +"backup version of packaging.py for porting" + +"Handles inkBoard packages, both creating and installing them." + +import asyncio +import zipfile +import os +import tempfile +import shutil +import inspect +import json +import subprocess +import sys + +from typing import TYPE_CHECKING, TypedDict, Literal, Callable, Union, Optional +from abc import abstractmethod +from functools import partial +from pathlib import Path +from datetime import datetime as dt +from contextlib import suppress + +import inkBoard +# import inkBoard.platforms +from inkBoard.configuration.const import CONFIG_FILE_TYPES, INKBOARD_FOLDER +from inkBoard.types import * +from inkBoard import constants as const, bootstrap + +import PythonScreenStackManager as PSSM + +from .constants import ( + ZIP_COMPRESSION, + ZIP_COMPRESSION_LEVEL, + ) + +if TYPE_CHECKING: + from inkBoard import CORE as CORE + from packaging.version import Version + +with suppress(ModuleNotFoundError): + import inkBoarddesigner + +try: + from packaging.version import parse as parse_version +except ModuleNotFoundError: + from pkg_resources import parse_version + + + +_LOGGER = inkBoard.getLogger(__name__) + +packagetypes = Literal['package', 'integration', 'platform'] +packageidfiles : dict[packagetypes,str] = { + 'package': 'package.json', + 'integration': 'manifest.json', + 'platform': 'platform.json' +} +internalinstalltypes = Literal["platform", "integration"] + +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 NegativeConfirmation(UserWarning): + "Raised by ask confirm if the confirmation was negative" + +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. + +VERSION_COMPARITORS = ('==', '!=', '>=', '<=', '>', '<') +"Comparison operators allowed for versioning, so they can be evaluated internally" + +DESIGNER_FILES = {"designer", "designer.py"} + +REQUIREMENTS_FILE = 'requirements.txt' + +required_attributes = { + "config", + "device", + "integration_loader" + +} + +InkboardVersion = parse_version(inkBoard.__version__) +PSSMVersion = parse_version(PSSM.__version__) + +def get_comparitor_string(input_str: str) -> Literal[VERSION_COMPARITORS]: + "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 confirm_input(msg: str, installer: "BaseInstaller"): + 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 + """ + + 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): + 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 + 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 + + +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(f"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 = self.CORE.integrationLoader.imported_integrations + + _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: + + package_dict = {"created": dt.now().isoformat()} + + package_dict["versions"] = {"inkBoard": inkBoard.__version__, + "PythonScreenStackManager": PSSM.__version__} + + + if self.CORE.DESIGNER_RUN: + 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) + +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 / packageidfiles["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 / packageidfiles["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(packageidfiles["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 / packageidfiles["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 + + # with self.__zip_file.open(f"{platform_info.filename}{packageidfiles['platform']}") as f: + f = self.__zip_file.open(f"{platform_info.filename}{packageidfiles['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 / packageidfiles["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 / packageidfiles['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 / packageidfiles["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] / packageidfiles["integration"]).exists(): + return 'integration' + elif (root_files[0] / packageidfiles["platform"]).exists(): + return 'platform' + elif (p / packageidfiles["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 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 const.DESIGNER_INSTALLED: + assert full_path.exists(), f"{install_type} {name} is not installed or does not exist" + else: + if not full_path.exists(): + assert (const.DESIGNER_FOLDER / file).exists(), f"{install_type} {name} is not installed or does not exist" + full_path = const.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 / packageidfiles["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 / packageidfiles["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 From da32d87482aa97413d2e70a7c235434bceb3f43c Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:02:38 +0200 Subject: [PATCH 05/18] port packager --- inkBoard/packaging/package.py | 264 ++++++++++++++++++++++++++++++++++ 1 file changed, 264 insertions(+) create mode 100644 inkBoard/packaging/package.py 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) From a141901043d1b2121201d19b41f8ab1c79e2ed47 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:03:32 +0200 Subject: [PATCH 06/18] clean up unused imports --- inkBoard/packaging/__init__.py | 351 ++------------------------------- 1 file changed, 21 insertions(+), 330 deletions(-) diff --git a/inkBoard/packaging/__init__.py b/inkBoard/packaging/__init__.py index 4d181de..29823e9 100644 --- a/inkBoard/packaging/__init__.py +++ b/inkBoard/packaging/__init__.py @@ -5,16 +5,13 @@ import os import tempfile import shutil -import inspect import json import subprocess import sys -from typing import TYPE_CHECKING, TypedDict, Literal, Callable, Union, Optional +from typing import TYPE_CHECKING, Callable, Union, Optional from abc import abstractmethod -from functools import partial from pathlib import Path -from datetime import datetime as dt from contextlib import suppress import inkBoard @@ -23,61 +20,22 @@ from inkBoard.types import * from inkBoard import constants as const, bootstrap -import PythonScreenStackManager as PSSM - +from .constants import ( + INKBOARD_PACKAGE_INTERNAL_FOLDER, + ) +from .types import ( + internalinstalltypes, + packagetypes, +) if TYPE_CHECKING: from inkBoard import CORE as CORE - from packaging.version import Version - -with suppress(ModuleNotFoundError): - import inkBoarddesigner - -try: - from packaging.version import parse as parse_version -except ModuleNotFoundError: - from pkg_resources import parse_version - -ZIP_COMPRESSION = zipfile.ZIP_BZIP2 -ZIP_COMPRESSION_LEVEL = 9 _LOGGER = inkBoard.getLogger(__name__) -packagetypes = Literal['package', 'integration', 'platform'] -packageidfiles : dict[packagetypes,str] = { - 'package': 'package.json', - 'integration': 'manifest.json', - 'platform': 'platform.json' -} -internalinstalltypes = Literal["platform", "integration"] - -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 NegativeConfirmation(UserWarning): "Raised by ask confirm if the confirmation was negative" -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. - -VERSION_COMPARITORS = ('==', '!=', '>=', '<=', '>', '<') -"Comparison operators allowed for versioning, so they can be evaluated internally" - -DESIGNER_FILES = {"designer", "designer.py"} - -REQUIREMENTS_FILE = 'requirements.txt' - required_attributes = { "config", "device", @@ -85,49 +43,6 @@ class NegativeConfirmation(UserWarning): } -InkboardVersion = parse_version(inkBoard.__version__) -PSSMVersion = parse_version(PSSM.__version__) - -def get_comparitor_string(input_str: str) -> Literal[VERSION_COMPARITORS]: - "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 = f"compare_version >= required_version" - - return eval(comp_str, {}, {"compare_version": compare_version, "required_version": parse_version(req_version)}) - def confirm_input(msg: str, installer: "BaseInstaller"): answer = input(f"{msg}\n(Y/N): ") if answer.lower() in {"y","yes"}: @@ -219,230 +134,6 @@ def install_packages(file: Union[str, Path] = None, no_input: bool = False): return 0 -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(f"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 = self.CORE.integrationLoader.imported_integrations - - _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: - - package_dict = {"created": dt.now().isoformat()} - - package_dict["versions"] = {"inkBoard": inkBoard.__version__, - "PythonScreenStackManager": PSSM.__version__} - - - if self.CORE.DESIGNER_RUN: - 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) class BaseInstaller: """Base class for installers @@ -610,7 +301,7 @@ def check_inkboard_requirements(self, ib_requirements: inkboardrequirements, req ##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 / packageidfiles["platform"]) as f: + with open(INKBOARD_FOLDER / "platforms" / platform / PACKAGE_ID_FILES["platform"]) as f: platform_conf: platformjson = json.load(f) cur_version = platform_conf["version"] @@ -630,7 +321,7 @@ def check_inkboard_requirements(self, ib_requirements: inkboardrequirements, req ##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 / packageidfiles["integration"]) as f: + with open(INKBOARD_FOLDER / "integrations" / integration / PACKAGE_ID_FILES["integration"]) as f: integration_conf: manifestjson = json.load(f) cur_version = integration_conf["version"] @@ -764,7 +455,7 @@ def install_package(self) -> Optional[packagetypes]: 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(packageidfiles["package"]) + f = zip_file.open(PACKAGE_ID_FILES["package"]) package_info: PackageDict = json.load(f) vers_msg = "" @@ -911,7 +602,7 @@ def install_config_requirements(self, config_file: Union[str,Path]): if (folder / "integrations").exists(): for integration_folder in (folder / "integrations").iterdir(): - with open(integration_folder / packageidfiles["integration"]) as f: + with open(integration_folder / PACKAGE_ID_FILES["integration"]) as f: integration_conf: manifestjson = json.load(f) if reqs := integration_conf.get("requirements", []): @@ -939,7 +630,7 @@ def _install_platform_zipinfo(self, platform_info: zipfile.ZipInfo): platform = platform_zippath.name # with self.__zip_file.open(f"{platform_info.filename}{packageidfiles['platform']}") as f: - f = self.__zip_file.open(f"{platform_info.filename}{packageidfiles['platform']}") + 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']) @@ -952,7 +643,7 @@ def _install_platform_zipinfo(self, platform_info: zipfile.ZipInfo): if (INKBOARD_FOLDER / "platforms" / platform).exists(): - with open(INKBOARD_FOLDER / "platforms" / platform / packageidfiles["platform"]) as f: + 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']) @@ -983,7 +674,7 @@ def _install_integration_zipinfo(self, integration_info: zipfile.ZipInfo): integration_zippath = zipfile.Path(self.__zip_file, integration_info.filename) integration = integration_zippath.name - manifestpath = integration_zippath / packageidfiles['integration'] + manifestpath = integration_zippath / PACKAGE_ID_FILES['integration'] f = manifestpath.open() integration_conf: manifestjson = json.load(f) integration_version = parse_version(integration_conf['version']) @@ -997,7 +688,7 @@ def _install_integration_zipinfo(self, integration_info: zipfile.ZipInfo): if (INKBOARD_FOLDER / "integrations" / integration).exists(): - with open(INKBOARD_FOLDER / "integrations" / integration / packageidfiles["integration"]) as f: + 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']) @@ -1122,11 +813,11 @@ def identify_zip_file(cls, file: Union[str, Path, zipfile.ZipFile]) -> Optional[ 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] / packageidfiles["integration"]).exists(): + if (root_files[0] / PACKAGE_ID_FILES["integration"]).exists(): return 'integration' - elif (root_files[0] / packageidfiles["platform"]).exists(): + elif (root_files[0] / PACKAGE_ID_FILES["platform"]).exists(): return 'platform' - elif (p / packageidfiles["package"]).exists() and (len(root_files) in {2,3}): ##2 or 3: at least contains package.json, and has .inkBoard and/or configuration folder + 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 @@ -1195,7 +886,7 @@ def install(self): def install_platform(self): - with open(self._full_path / packageidfiles["platform"]) as f: + with open(self._full_path / PACKAGE_ID_FILES["platform"]) as f: conf: platformjson = json.load(f) with suppress(NegativeConfirmation): @@ -1205,7 +896,7 @@ def install_platform(self): return 1 def install_integration(self): - with open(self._full_path / packageidfiles["integration"]) as f: + with open(self._full_path / PACKAGE_ID_FILES["integration"]) as f: conf: platformjson = json.load(f) with suppress(NegativeConfirmation): From c26f84e25c04096106696d0655049fdc724ac314 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:16:00 +0200 Subject: [PATCH 07/18] port installer code --- inkBoard/packaging/install.py | 819 ++++++++++++++++++++++++++++++++++ 1 file changed, 819 insertions(+) create mode 100644 inkBoard/packaging/install.py 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 From 51bf160e6db771dde74b1bfc2971b6be41ba6937 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:16:17 +0200 Subject: [PATCH 08/18] port NegativeConfirmation to prevent circular import --- inkBoard/packaging/types.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/inkBoard/packaging/types.py b/inkBoard/packaging/types.py index 3515cc5..dd6c9c7 100644 --- a/inkBoard/packaging/types.py +++ b/inkBoard/packaging/types.py @@ -40,4 +40,7 @@ class PackageIndex(TypedDict): comparisonstrings = Literal[ '==', '!=', '>=', '<=', '>', '<' -] \ No newline at end of file +] + +class NegativeConfirmation(UserWarning): + "Raised by ask confirm if the confirmation was negative" \ No newline at end of file From be74b5e7cedc6c7319282bd6278c502056b067ef Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:19:46 +0200 Subject: [PATCH 09/18] cleanup unused imports and single use ones --- inkBoard/packaging/__init__.py | 810 +-------------------------------- 1 file changed, 8 insertions(+), 802 deletions(-) diff --git a/inkBoard/packaging/__init__.py b/inkBoard/packaging/__init__.py index 29823e9..5e1c939 100644 --- a/inkBoard/packaging/__init__.py +++ b/inkBoard/packaging/__init__.py @@ -1,48 +1,21 @@ "Handles inkBoard packages, both creating and installing them." import asyncio -import zipfile -import os -import tempfile -import shutil -import json -import subprocess -import sys - -from typing import TYPE_CHECKING, Callable, Union, Optional -from abc import abstractmethod +from typing import TYPE_CHECKING, Union from pathlib import Path -from contextlib import suppress import inkBoard -# import inkBoard.platforms -from inkBoard.configuration.const import CONFIG_FILE_TYPES, INKBOARD_FOLDER -from inkBoard.types import * -from inkBoard import constants as const, bootstrap +# from inkBoard.types import * +from inkBoard import constants as bootstrap +from .types import internalinstalltypes -from .constants import ( - INKBOARD_PACKAGE_INTERNAL_FOLDER, - ) -from .types import ( - internalinstalltypes, - packagetypes, -) if TYPE_CHECKING: from inkBoard import CORE as CORE + from .install import BaseInstaller _LOGGER = inkBoard.getLogger(__name__) -class NegativeConfirmation(UserWarning): - "Raised by ask confirm if the confirmation was negative" - -required_attributes = { - "config", - "device", - "integration_loader" - -} - def confirm_input(msg: str, installer: "BaseInstaller"): answer = input(f"{msg}\n(Y/N): ") if answer.lower() in {"y","yes"}: @@ -89,7 +62,7 @@ def create_core_package(core: "CORE", name: str = None, pack_all: bool = False, core : CORE The core object constructed from the config """ - + from .package import Packager if pack_all: Packager(core).create_package(name) else: @@ -112,6 +85,7 @@ def run_install_command(file: str, name: str, no_input: bool): 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): @@ -122,6 +96,7 @@ def install_packages(file: Union[str, Path] = None, no_input: bool = False): ##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: @@ -135,772 +110,3 @@ def install_packages(file: Union[str, Path] = None, no_input: bool = False): -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 - - # with self.__zip_file.open(f"{platform_info.filename}{packageidfiles['platform']}") as f: - 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 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 const.DESIGNER_INSTALLED: - assert full_path.exists(), f"{install_type} {name} is not installed or does not exist" - else: - if not full_path.exists(): - assert (const.DESIGNER_FOLDER / file).exists(), f"{install_type} {name} is not installed or does not exist" - full_path = const.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 From 96354d3648becaae1d856cfb1c4f0b6072285e1a Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:29:17 +0200 Subject: [PATCH 10/18] fix a grammar error --- inkBoard/arguments.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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" From 6ad8d2c48b14f906d9c59bd8236c236b77d328b5 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:31:58 +0200 Subject: [PATCH 11/18] add log message for myself --- inkBoard/packaging/__init__.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/inkBoard/packaging/__init__.py b/inkBoard/packaging/__init__.py index 5e1c939..508605d 100644 --- a/inkBoard/packaging/__init__.py +++ b/inkBoard/packaging/__init__.py @@ -12,11 +12,11 @@ if TYPE_CHECKING: from inkBoard import CORE as CORE - from .install import BaseInstaller _LOGGER = inkBoard.getLogger(__name__) +_LOGGER.warning("Dont forget to write tests fro this") -def confirm_input(msg: str, installer: "BaseInstaller"): +def confirm_input(msg: str): answer = input(f"{msg}\n(Y/N): ") if answer.lower() in {"y","yes"}: return True @@ -63,6 +63,7 @@ def create_core_package(core: "CORE", name: str = None, pack_all: bool = False, The core object constructed from the config """ from .package import Packager + if pack_all: Packager(core).create_package(name) else: From bc1f01b373990ab8a4fcf84fc1548c07c43f3461 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 16:32:59 +0200 Subject: [PATCH 12/18] start writing downloader --- inkBoard/packaging/download.py | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 inkBoard/packaging/download.py diff --git a/inkBoard/packaging/download.py b/inkBoard/packaging/download.py new file mode 100644 index 0000000..010cbc1 --- /dev/null +++ b/inkBoard/packaging/download.py @@ -0,0 +1,3 @@ +"""Functions to download stuff from the package index +Trying to implement this without using requests +""" \ No newline at end of file From 7a215a8671d95ba3c9f730fc82c115316e159411 Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 21:20:36 +0200 Subject: [PATCH 13/18] add constant for the name of the index file internally --- inkBoard/packaging/constants.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/inkBoard/packaging/constants.py b/inkBoard/packaging/constants.py index 0f6befe..30c9ca1 100644 --- a/inkBoard/packaging/constants.py +++ b/inkBoard/packaging/constants.py @@ -12,6 +12,8 @@ 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 From 0a58d453f38d153e23a5297ff9fc8761b56b1abc Mon Sep 17 00:00:00 2001 From: Sander Date: Sun, 22 Jun 2025 21:21:35 +0200 Subject: [PATCH 14/18] =?UTF-8?q?feat(packaging):=20=E2=9C=A8=20base=20dow?= =?UTF-8?q?nloader=20functionality?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Can now get the index.json file from the package index. Should be extendable to download the zips too. --- inkBoard/packaging/download.py | 100 ++++++++++++++++++++++++++++++++- 1 file changed, 99 insertions(+), 1 deletion(-) diff --git a/inkBoard/packaging/download.py b/inkBoard/packaging/download.py index 010cbc1..df0e08d 100644 --- a/inkBoard/packaging/download.py +++ b/inkBoard/packaging/download.py @@ -1,3 +1,101 @@ """Functions to download stuff from the package index Trying to implement this without using requests -""" \ No newline at end of file +""" +from typing import ( + Union, +) +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, +) + +_LOGGER = logging.getLogger(__name__) + +##See this repo: https://github.com/fbunaren/GitHubFolderDownloader +##And this gist: https://gist.github.com/oculushut/193a7c2b6002d808a791 + +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" + + def __init__(self, destination_folder : Path): + 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 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() + + with open(internal_file) as f: + package_index = PackageIndex(json.load(f)) + + self.index : PackageIndex = package_index + return package_index + + @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 + raw_url = cls._make_raw_file_link(PACKAGE_INDEX_URL, "index.json") + cls._download_raw_file(raw_url, _destination_file) + + @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(repo_url : str, file : str, branch = "main"): + """Returns the link to the raw file on github + + Parameters + ---------- + repo_url : str + url to the repo + file : str + _description_ + branch : str, optional + _description_, by default "main" + + Returns + ------- + _type_ + _description_ + """ + 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 From be2d5d2abf2477cd7f7d0a6551fc559ad20bc870 Mon Sep 17 00:00:00 2001 From: Sander Date: Tue, 24 Jun 2025 09:14:26 +0200 Subject: [PATCH 15/18] Add package index file to ignoreables --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) 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 From a4711db39e51dd1cd628b6875836e38b19629e19 Mon Sep 17 00:00:00 2001 From: Sander Date: Tue, 24 Jun 2025 09:15:50 +0200 Subject: [PATCH 16/18] =?UTF-8?q?feat(packaging):=20=F0=9F=9A=A7=20Get=20s?= =?UTF-8?q?tarted=20with=20downloading=20integration=20packages?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- inkBoard/packaging/download.py | 74 ++++++++++++++++++++++++++++++---- 1 file changed, 66 insertions(+), 8 deletions(-) diff --git a/inkBoard/packaging/download.py b/inkBoard/packaging/download.py index df0e08d..314320d 100644 --- a/inkBoard/packaging/download.py +++ b/inkBoard/packaging/download.py @@ -3,6 +3,8 @@ """ from typing import ( Union, + Literal, + ) import urllib.request from pathlib import Path @@ -22,10 +24,14 @@ ) _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 @@ -34,7 +40,16 @@ class Downloader: 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: @@ -45,6 +60,8 @@ def get_package_index(self, force_get = False) -> dict: 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) @@ -56,6 +73,7 @@ def get_package_index(self, force_get = False) -> dict: if get_index: self._download_package_index() + self._index_downloaded = True with open(internal_file) as f: package_index = PackageIndex(json.load(f)) @@ -63,12 +81,52 @@ def get_package_index(self, force_get = False) -> dict: 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() + 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 - raw_url = cls._make_raw_file_link(PACKAGE_INDEX_URL, "index.json") + _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): @@ -78,22 +136,22 @@ def _download_raw_file(raw_file_url : str, destination_file : str): return filename, headers @staticmethod - def _make_raw_file_link(repo_url : str, file : str, branch = "main"): + 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 ---------- - repo_url : str - url to the repo file : str - _description_ + The path to the file to download branch : str, optional - _description_, by default "main" + 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 ------- - _type_ - _description_ + 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("/") From b5961d47af783938e873110c578f50a097696557 Mon Sep 17 00:00:00 2001 From: Sander Date: Mon, 7 Jul 2025 21:46:58 +0200 Subject: [PATCH 17/18] =?UTF-8?q?fix:=20=F0=9F=A9=B9=20Fix=20log=20format?= =?UTF-8?q?=20accidentally=20being=20hardcoded?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bandaid fix, needs a better one that is more in Python style --- inkBoard/logging.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) 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: From fa16aed6ca1d75ce58ac6370f0c4728a0ebc9d64 Mon Sep 17 00:00:00 2001 From: Sander Date: Tue, 8 Jul 2025 10:09:22 +0200 Subject: [PATCH 18/18] feat: improve structure and packaging --- inkBoard/packaging/download.py | 7 ++++++- inkBoard/packaging/version.py | 26 ++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/inkBoard/packaging/download.py b/inkBoard/packaging/download.py index 314320d..b863d51 100644 --- a/inkBoard/packaging/download.py +++ b/inkBoard/packaging/download.py @@ -4,7 +4,7 @@ from typing import ( Union, Literal, - + TYPE_CHECKING ) import urllib.request from pathlib import Path @@ -23,6 +23,9 @@ 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 @@ -116,6 +119,8 @@ def download_integration_package(self, name : str, package_type : Literal["main" pass raw_url = self._make_raw_file_link() + + #[ ] for platforms, ask to copy files like readme etc. into the current working directory? return @classmethod diff --git a/inkBoard/packaging/version.py b/inkBoard/packaging/version.py index a8dc52d..5392101 100644 --- a/inkBoard/packaging/version.py +++ b/inkBoard/packaging/version.py @@ -63,3 +63,29 @@ def compare_versions(requirement: Union[str,"Version"], compare_version: Union[s 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