Automated Conda Environment Builder Tool Project 02.12.2024.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Project Workflow: Automated Conda Environment Builder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "## Define Detailed Steps:\n",
    "\n",
    "### Search for Pre-existing Packages:\n",
    "Leverage existing packages to avoid reinventing the wheel. Look for packages that meet the project requirements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to search for pre-existing packages in a conda environment and install them if they are not present\n",
    "import subprocess\n",
    "import json\n",
    "\n",
    "def check_package_installed(env_name, package):\n",
    "    result = subprocess.run([\"conda\", \"list\", \"--name\", env_name, package, \"--json\"], stdout=subprocess.PIPE, check=True)\n",
    "    packages = json.loads(result.stdout)\n",
    "    return any(pkg['name'] == package for pkg in packages)\n",
    "\n",
    "def create_conda_env(env_name, packages):\n",
    "    # Create a new conda environment and install packages\n",
    "    commands = [\n",
    "        \"conda create --name {} --yes\".format(env_name),\n",
    "    ] + [\n",
    "        \"conda install --name {} --yes {}\".format(env_name, package) if not check_package_installed(env_name, package) else \"\"\n",
    "        for package in packages\n",
    "    ]\n",
    "\n",
    "    script = \" && \".join(cmd for cmd in commands if cmd)\n",
    "\n",
    "    subprocess.run(script, shell=True, check=True)\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Select the Best-Fitting Package:\n",
    "Choose an existing package that aligns with the initial requirements. Consider factors like popularity, community support, and reliability."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to select the best-fitting package\n",
    "def create_conda_env(env_name, packages):\n",
    "    # Create a new conda environment\n",
    "    subprocess.run([\"conda\", \"create\", \"--name\", env_name, \"--yes\"], check=True)\n",
    "\n",
    "    # Install packages\n",
    "    for package in packages:\n",
    "        name, version = package.split('==') if '==' in package else (package, \"\")\n",
    "        if not check_package_installed(env_name, name):\n",
    "            subprocess.run([\"conda\", \"install\", \"--name\", env_name, \"--yes\", package], check=True)\n",
    "\n",
    "    # Activate the environment\n",
    "    subprocess.run([\"conda\", \"activate\", env_name], check=True)\n",
    "\n",
    "# Usage\n",
    "create_conda_env(\"my_env\", [\"numpy==1.18.5\", \"pandas\", \"scikit-learn\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Download the Best-Fitting Package:\n",
    "Retrieve the selected package. Ensure it is readily available for use."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Code to download the best-fitting package\n",
    "import re\n",
    "\n",
    "def get_latest_package_version(package):\n",
    "    result = subprocess.run([\"conda\", \"search\", \"--json\", package], stdout=subprocess.PIPE, check=True)\n",
    "    packages = json.loads(result.stdout)\n",
    "    versions = [pkg['version'] for pkg in packages[package]]\n",
    "    versions.sort(key=lambda s: list(map(int, re.findall(r'\\d+', s))), reverse=True)\n",
    "    return versions[0] if versions else None\n",
    "\n",
    "def create_conda_env(env_name, packages):\n",
    "    # Create a new conda environment\n",
    "    subprocess.run([\"conda\", \"create\", \"--name\", env_name, \"--yes\"], check=True)\n",
    "\n",
    "    # Install packages\n",
    "    for package in packages:\n",
    "        name, version = package.split('==') if '==' in package else (package, \"\")\n",
    "        if not version:\n",
    "            version = get_latest_package_version(name)\n",
    "        if not check_package_installed(env_name, name):\n",
    "            subprocess.run([\"conda\", \"install\", \"--name\", env_name, \"--yes\", f\"{name}=={version}\"], check=True)\n",
    "\n",
    "    # Activate the environment\n",
    "    subprocess.run([\"conda\", \"activate\", env_name], check=True)\n",
    "\n",
    "# Usage\n",
    "create_conda_env(\"my_env\", [\"numpy\", \"pandas\", \"scikit-learn\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### User-Specify Project Folder Name:\n",
    "Create a function (get_new_project_name()) to prompt the user for a project folder name. Use this name for the new project directory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_new_project_name():\n",
    "    # Prompt the user for a project folder name\n",
    "import os\n",
    "\n",
    "def get_new_project_name():\n",
    "    while True:\n",
    "        project_name = input(\"Enter a name for the new project directory: \")\n",
    "        if not os.path.exists(project_name):\n",
    "            os.makedirs(project_name)\n",
    "            print(f\"Created new project directory: {project_name}\")\n",
    "            break\n",
    "        else:\n",
    "            print(\"A directory with this name already exists. Please enter a different name.\")\n",
    "\n",
    "# Usage\n",
    "get_new_project_name()   pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Unzip the Best-Fitting Package into New Project Folder:\n",
    "Implement a function (unzip_starter_project_repo()) to extract the package contents. Place them in the user-specified project folder."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to extract the package contents into the user-specified project folder\n",
    "\n",
    "def unzip_starter_project_repo():\n",
    "import zipfile\n",
    "\n",
    "def unzip_starter_project_repo(project_name):\n",
    "    with zipfile.ZipFile('best_fitting_package.zip', 'r') as zip_ref:\n",
    "        zip_ref.extractall(project_name)\n",
    "\n",
    "# Usage\n",
    "# Assume that the project_name is obtained from the get_new_project_name() function\n",
    "project_name = get_new_project_name()\n",
    "unzip_starter_project_repo(project_name)    \n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Scan Project Folder for Required Python Package Dependencies:\n",
    "Develop a function (scan_project_modules()) to identify the unique set of required Python packages. Store these dependencies in a set (py_dependencies_set)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Scan Project Folder for Required Python Package Dependencies:\n",
    "import os\n",
    "import re\n",
    "\n",
    "def scan_project_modules(project_name):\n",
    "    py_dependencies_set = set()\n",
    "\n",
    "    for root, dirs, files in os.walk(project_name):\n",
    "        for file in files:\n",
    "            if file.endswith('.py'):\n",
    "                with open(os.path.join(root, file), 'r') as f:\n",
    "                    content = f.read()\n",
    "                    imports = re.findall(r'^import\\s+(\\w+)', content, re.MULTILINE)\n",
    "                    from_imports = re.findall(r'^from\\s+(\\w+)', content, re.MULTILINE)\n",
    "                    py_dependencies_set.update(imports, from_imports)\n",
    "\n",
    "    return py_dependencies_set\n",
    "\n",
    "# Usage\n",
    "# Assume that the project_name is obtained from the get_new_project_name() function\n",
    "project_name = get_new_project_name()\n",
    "py_dependencies_set = scan_project_modules(project_name)\n",
    "print(py_dependencies_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Scan Local Pre-existing Virtual Environments:\n",
    "Create a function (scan_existing_env()) to analyze our existing virtual environments. Determine which environment best fits the dependency tree (current_env_dependencies_set)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Scan Local Pre-existing Virtual Environments:\n",
    "def scan_existing_env(current_env_dependencies_set):\n",
    "    result = subprocess.run([\"conda\", \"env\", \"list\"], stdout=subprocess.PIPE, check=True)\n",
    "    envs = result.stdout.decode().splitlines()[3:]  # Skip the header lines\n",
    "\n",
    "    best_env = None\n",
    "    best_match_count = 0\n",
    "\n",
    "    for env in envs:\n",
    "        env_name = env.split()[0]\n",
    "        result = subprocess.run([\"conda\", \"list\", \"--name\", env_name, \"--json\"], stdout=subprocess.PIPE, check=True)\n",
    "        packages = json.loads(result.stdout)\n",
    "        package_names = {pkg['name'] for pkg in packages}\n",
    "\n",
    "        match_count = len(current_env_dependencies_set & package_names)\n",
    "        if match_count > best_match_count:\n",
    "            best_env = env_name\n",
    "            best_match_count = match_count\n",
    "\n",
    "    return best_env\n",
    "\n",
    "# Usage\n",
    "# Assume that current_env_dependencies_set is obtained from the scan_project_modules() function\n",
    "current_env_dependencies_set = scan_project_modules(project_name)\n",
    "best_env = scan_existing_env(current_env_dependencies_set)\n",
    "print(best_env)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Decide on Environment Strategy:\n",
    "Evaluate whether to use or clone an existing virtual environment or create a new one. Consider whether existing environments have the required up-to-date packages."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to decide on environment strategy\n",
    "def decide_environment_strategy(project_name):\n",
    "    # Get the set of dependencies of the project\n",
    "    project_dependencies = scan_project_modules(project_name)\n",
    "\n",
    "    # Get the best fitting existing environment\n",
    "    best_env = scan_existing_env(project_dependencies)\n",
    "\n",
    "    if best_env is not None:\n",
    "        # Get the set of packages in the best fitting environment\n",
    "        result = subprocess.run([\"conda\", \"list\", \"--name\", best_env, \"--json\"], stdout=subprocess.PIPE, check=True)\n",
    "        packages = json.loads(result.stdout)\n",
    "        best_env_packages = {pkg['name'] for pkg in packages}\n",
    "\n",
    "        # Check if the best fitting environment contains all the dependencies of the project\n",
    "        if project_dependencies.issubset(best_env_packages):\n",
    "            print(f\"The existing environment '{best_env}' can be used directly.\")\n",
    "        else:\n",
    "            missing_packages = project_dependencies - best_env_packages\n",
    "            print(f\"The existing environment '{best_env}' can be cloned and the following packages can be installed: {missing_packages}\")\n",
    "    else:\n",
    "        print(\"A new environment can be created.\")\n",
    "\n",
    "# Usage\n",
    "# Assume that the project_name is obtained from the get_new_project_name() function\n",
    "project_name = get_new_project_name()\n",
    "decide_environment_strategy(project_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Decide on Environment Strategy:\n",
    "Evaluate whether to use or clone an existing virtual environment or create a new one. Consider whether existing environments have the required up-to-date packages."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Decide on Environment Strategy:\n",
    "\n",
    "import subprocess\n",
    "import json\n",
    "\n",
    "def decide_environment_strategy(project_dependencies):\n",
    "    # Get the best fitting existing environment\n",
    "    best_env = scan_existing_env(project_dependencies)\n",
    "\n",
    "    if best_env is not None:\n",
    "        # Get the set of packages in the best fitting environment\n",
    "        result = subprocess.run([\"conda\", \"list\", \"--name\", best_env, \"--json\"], stdout=subprocess.PIPE, check=True)\n",
    "        packages = json.loads(result.stdout)\n",
    "        best_env_packages = {pkg['name'] for pkg in packages}\n",
    "\n",
    "        # Check if the best fitting environment contains all the dependencies of the project\n",
    "        if project_dependencies.issubset(best_env_packages):\n",
    "            print(f\"The existing environment '{best_env}' can be used directly.\")\n",
    "            return best_env, 'use'\n",
    "        else:\n",
    "            missing_packages = project_dependencies - best_env_packages\n",
    "            print(f\"The existing environment '{best_env}' can be cloned and the following packages can be installed: {missing_packages}\")\n",
    "            return best_env, 'clone'\n",
    "    else:\n",
    "        print(\"A new environment can be created.\")\n",
    "        return None, 'create'\n",
    "\n",
    "# Usage\n",
    "# Assume that project_dependencies is obtained from the scan_project_modules() function\n",
    "project_dependencies = scan_project_modules(project_name)\n",
    "best_env, strategy = decide_environment_strategy(project_dependencies)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Iterate Until All Packages and Versions Are Found:\n",
    "Continuously search and verify package availability. Build lists of repositories (conda-forge, anaconda, CONDA-OTHER, or pip) containing required versions. Ensure all dependencies are covered."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to iterate until all packages and versions are found\n",
    "\n",
    "import subprocess\n",
    "import json\n",
    "\n",
    "def search_packages(package_versions):\n",
    "    repositories = ['conda-forge', 'anaconda']\n",
    "    available_packages = {repo: [] for repo in repositories}\n",
    "\n",
    "    for package, version in package_versions.items():\n",
    "        for repo in repositories:\n",
    "            command = [\"conda\", \"search\", \"--channel\", repo, f\"{package}={version}\", \"--json\"]\n",
    "            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n",
    "\n",
    "            if result.returncode == 0:\n",
    "                data = json.loads(result.stdout)\n",
    "                if data:\n",
    "                    available_packages[repo].append(package)\n",
    "\n",
    "    return available_packages\n",
    "\n",
    "# Usage\n",
    "# Assume that package_versions is a dictionary where the keys are package names and the values are the required versions\n",
    "package_versions = {\"numpy\": \"1.21.0\", \"pandas\": \"1.3.0\"}\n",
    "available_packages = search_packages(package_versions)\n",
    "print(available_packages)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Create Python Functions for Environment Execution:\n",
    "Develop functions or separate script commands needed to set up and activate the environment. Handle installation of packages from the identified repositories."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code to create Python functions for environment execution\n",
    "import subprocess\n",
    "\n",
    "def create_and_setup_environment(env_name, packages, repo='conda-forge'):\n",
    "    # Create a new environment\n",
    "    result = subprocess.run([\"conda\", \"create\", \"--name\", env_name, \"--yes\"], check=True)\n",
    "    if result.returncode != 0:\n",
    "        print(f\"Failed to create environment '{env_name}'.\")\n",
    "        return\n",
    "\n",
    "    # Activate the environment\n",
    "    # Note: This won't affect the parent shell environment, so it won't actually switch the active Conda environment.\n",
    "    # In a real use case, you would need to instruct the user to activate the environment manually.\n",
    "    result = subprocess.run([\"conda\", \"activate\", env_name], check=True)\n",
    "    if result.returncode != 0:\n",
    "        print(f\"Failed to activate environment '{env_name}'.\")\n",
    "        return\n",
    "\n",
    "    # Install packages\n",
    "    for package in packages:\n",
    "        result = subprocess.run([\"conda\", \"install\", \"--channel\", repo, package, \"--yes\"], check=True)\n",
    "        if result.returncode != 0:\n",
    "            print(f\"Failed to install package '{package}'.\")\n",
    "\n",
    "# Usage\n",
    "# Assume that env_name is the name of the new environment, and packages is a list of package names\n",
    "env_name = \"new_env\"\n",
    "packages = [\"numpy\", \"pandas\"]\n",
    "create_and_setup_environment(env_name, packages)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Create Python Functions (or Separate Script Commands) for Environment Build Steps:\n",
    "Implement the following steps as functions or separate script commands:\n",
    "- a) Create “Core” Environment: Use as many conda-forge packages as possible. Example: conda create -n new_env -c conda-forge python conda ...\n",
    "- b) Activate the New Environment: Use conda activate new_env.\n",
    "- c) Install Packages from Conda Channels: Install all other required packages from conda-forge, anaconda, or other specified channels. Example: conda install -c conda-forge package1 package2 ...\n",
    "- d) Install Remaining Packages via Pip: Install any remaining packages that are not available in conda format. Example: pip install package3 package4 ..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create Python Functions (or Separate Script Commands) for Environment Build Steps:\n",
    "\n",
    "import subprocess\n",
    "\n",
    "def create_core_environment(env_name):\n",
    "    # Create a new environment with python and conda from conda-forge\n",
    "    result = subprocess.run([\"conda\", \"create\", \"--name\", env_name, \"-c\", \"conda-forge\", \"python\", \"conda\", \"--yes\"], check=True)\n",
    "    if result.returncode != 0:\n",
    "        print(f\"Failed to create environment '{env_name}'.\")\n",
    "        return False\n",
    "    return True\n",
    "\n",
    "def activate_environment(env_name):\n",
    "    # Note: This won't affect the parent shell environment, so it won't actually switch the active Conda environment.\n",
    "    # In a real use case, you would need to instruct the user to activate the environment manually.\n",
    "    result = subprocess.run([\"conda\", \"activate\", env_name], check=True)\n",
    "    if result.returncode != 0:\n",
    "        print(f\"Failed to activate environment '{env_name}'.\")\n",
    "        return False\n",
    "    return True\n",
    "\n",
    "def install_packages_from_conda(env_name, packages, channel=\"conda-forge\"):\n",
    "    for package in packages:\n",
    "        result = subprocess.run([\"conda\", \"install\", \"--name\", env_name, \"-c\", channel, package, \"--yes\"], check=True)\n",
    "        if result.returncode != 0:\n",
    "            print(f\"Failed to install package '{package}' from channel '{channel}'.\")\n",
    "            return False\n",
    "    return True\n",
    "\n",
    "def install_remaining_packages_via_pip(env_name, packages):\n",
    "    for package in packages:\n",
    "        result = subprocess.run([\"conda\", \"run\", \"-n\", env_name, \"pip\", \"install\", package], check=True)\n",
    "        if result.returncode != 0:\n",
    "            print(f\"Failed to install package '{package}' via pip.\")\n",
    "            return False\n",
    "    return True\n",
    "\n",
    "# Usage\n",
    "env_name = \"new_env\"\n",
    "conda_packages = [\"package1\", \"package2\"]\n",
    "pip_packages = [\"package3\", \"package4\"]\n",
    "\n",
    "if create_core_environment(env_name):\n",
    "    if activate_environment(env_name):\n",
    "        if install_packages_from_conda(env_name, conda_packages):\n",
    "            install_remaining_packages_via_pip(env_name, pip_packages)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "markdown"
    }
   },
   "outputs": [],
   "source": [
    "### Build the Final Script to Execute the 12 Steps:\n",
    "Combine all the steps into a single script or workflow. Ensure proper error handling and logging."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "```python\n",
    "# Placeholder for code to build the final script to execute the"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pypowerbi",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}