import os.path import shutil import tempfile from pythonapp.Decider.ABS import SimpleDecider from pythonapp.Env.Venv import Venv, Env from pythonapp.Env.StandaloneEnv import StandalonePythonEnv from pythonapp.Libs.git import git from pythonapp.Decider.misc import * from pythonapp.Libs.pip_api import pip_api from .ABS import InstanceConfig, InstanceFileNaming, Instance as ABSInstance from ..Decider.Loader import Loader class Instance(ABSInstance): def __init__(self, path: str, env: str = 'venv', python = 'python3'): self.path = Path(path) self.config = InstanceConfig(os.path.join(self.path, InstanceFileNaming.main_config)) self.config.env_type = self.config.env_type or env self.config.created = (Path(self.path) / self.config.config_dir_rel_path).exists() if self.config.env_type == 'venv': self.env: Env = Venv(str(self.path / self.config.env_path), python) elif self.config.env_type == 'standalone': self.env: Env = StandalonePythonEnv(str(self.path / self.config.env_path), python) def insert_component_reqs(self, name: str, req_file: str | Path): try: req, opt = requirements_separator(req_file) os.makedirs(self.path / self.config.requirements_dir, exist_ok=True) with open(self.path / self.config.requirements_dir / (name + '.req'), 'w') as file: for line in req: file.write(line + '\n') with open(self.path / self.config.requirements_dir / (name + '.opt'), 'w') as file: for line in opt: file.write(line + '\n') except FileNotFoundError: raise RuntimeError(f"Cannot update requirements for {name}, file {req_file} not exists") def _symlink_move_dir(self, orig_dir: Path, dest: Path, obj_name: Path): cwd = os.getcwd() os.chdir(self.path) shutil.move(orig_dir / obj_name, dest) os.symlink(os.path.relpath(orig_dir, dest), orig_dir / obj_name) os.chdir(orig_dir) with open(".gitignore", "a") as file: file.write(str(obj_name) + '\n') os.chdir(cwd) def install_git_app(self, url: str, requirements_file_in_app_dir = 'requirements.txt', extensions_dir: str = None, models_dir: str = None, output_dir: str = None, input_dir: str = None, user_dir: str = None, ): if os.path.exists(self.config.app): raise RuntimeError("App installed previously. Multiapp instances is not supported") with tempfile.TemporaryDirectory() as tmp_dir: git.clone(url, tmp_dir) try: self.install_reqs('app', Path(tmp_dir) / requirements_file_in_app_dir) except RuntimeError as e: raise TypeError(e) self.config.app_extensions_dir = Path(extensions_dir) if extensions_dir else None self.config.app_models_dir = Path(models_dir) if models_dir else None self.config.app_output_dir = Path(output_dir) if output_dir else None self.config.app_input_dir = Path(input_dir) if input_dir else None self.config.app_user_dir = Path(user_dir) if user_dir else None self.config.save() git.clone(url, self.path / self.config.app) if self.config.app_extensions_dir: self._symlink_move_dir(Path(os.path.dirname(self.config.app / self.config.app_extensions_dir)), Path(os.path.basename(self.config.app / self.config.app_extensions_dir)), Path(os.path.basename(self.config.app / self.config.app_extensions_dir))) if self.config.app_models_dir: self._symlink_move_dir(Path(os.path.dirname(self.config.app / self.config.app_models_dir)), Path(os.path.basename(self.config.app / self.config.app_models_dir)), Path(os.path.basename(self.config.app / self.config.app_models_dir))) if self.config.app_output_dir: self._symlink_move_dir(Path(os.path.dirname(self.config.app / self.config.app_output_dir)), Path(os.path.basename(self.config.app / self.config.app_output_dir)), Path(os.path.basename(self.config.app / self.config.app_output_dir))) if self.config.app_input_dir: self._symlink_move_dir(Path(os.path.dirname(self.config.app / self.config.app_input_dir)), Path(os.path.basename(self.config.app / self.config.app_input_dir)), Path(os.path.basename(self.config.app / self.config.app_input_dir))) if self.config.app_user_dir: self._symlink_move_dir(Path(os.path.dirname(self.config.app / self.config.app_user_dir)), Path(os.path.basename(self.config.app / self.config.app_user_dir)), Path(os.path.basename(self.config.app / self.config.app_user_dir))) def install_reqs(self, name, req_file: Path): packages, errors, state = SimpleDecider.decide(self, req_file, self.env.pip_path) if not packages: raise RuntimeError("Cannot install packages due conflicts") self.env.install_pkgs([p.requirement_str for p in packages]) self.insert_component_reqs(name, req_file) def create(self): os.makedirs(self.path / self.config.config_dir_rel_path, exist_ok=True) self.config.save() self.env.create() def install_packages(self, pkgs: list, repo: str = None, extra: str = None, pin=False): packages = [RequirementInfo.from_requirement_string(p, 'manual', 'manual') for p in pkgs] packages, errors, state = SimpleDecider.decide(self, packages, self.env.pip_path, manual=True) pkgs = [p.requirement_str for p in packages] print("manually install packages", pkgs) self.env.install_pkgs(pkgs, repo, extra) for pkg in pkgs: self._write_unique_string(pkg, file_path=self.path / self.config.manual_requirements_path) if pin: for pkg in pkgs: self._write_unique_string(pkg, file_path=self.path / self.config.pinned_packages_path) def exclude_packages(self, pkgs: list[str]): for pkg in pkgs: self._write_unique_string(pkg, file_path=self.path / self.config.excluded_packages_path) def _write_unique_string(self, string, file_path): existing_lines = set() if file_path.exists(): with open(file_path, 'r') as file: existing_lines = set(line.strip() for line in file) with open(file_path, 'a') as file: if string not in existing_lines: file.write(string + '\n') def _delete_string(self, string, file_path): with open(file_path, 'r') as file: lines = file.readlines() filtered_lines = [line for line in lines if line.rstrip('\n') != string] with open(file_path, 'w') as file: file.writelines(filtered_lines)