Source code for cgl.plugins.unreal.utils.plugins
import logging
import os
# from cgl.core.cloud.auth import Cognito, handle_aws_login
from cgl.core.cloud.constants import cloud_constants
# from cgl.core.config.config import ProjectConfig
from cgl.core.config.query import get_perforce_root, AlchemyConfigManager
# from cgl.core.utils.network import check_for_network_drive_connection
from cgl.core.utils.read_write import load_yaml, load_json, unzip, zip_path
from cgl.plugins.unreal.utils.engine import (
get_unreal_marketplace_plugins_dir,
get_engine_version,
)
from cgl.plugins.unreal.utils.uproject import get_uproject_path
CFG = AlchemyConfigManager()
[docs]
def get_missing_local_plugins(project, location):
"""
Gets a list of plugins needed for the uproject that aren't downloaded to Unreal Engine
Args:
project (str): Project the location belongs to
location (str): Location to check for missing plugins
Returns:
missing_list (list): List of plugins that are missing from Unreal Engine
"""
missing_list = []
required_plugins = get_marketplace_plugins_for_location(
project=project, location=location
)
installed_plugins = get_installed_marketplace_plugins()
for plugin_name in required_plugins:
if plugin_name not in installed_plugins:
missing_list.append(plugin_name)
return missing_list
[docs]
def get_missing_s3_plugins(project, location):
"""
Gets a list of plugins needed for the uproject that aren't downloaded to Unreal Engine
Args:
project (str): Project the location belongs to
location (str): Location to check for missing plugins
Returns:
missing_list (list): List of plugins that are missing from Unreal Engine
"""
missing_list = []
required_plugins = get_marketplace_plugins_for_location(
project=project, location=location
)
engine_version = get_engine_version()
uploaded_plugins = get_plugins_in_s3(engine_version=engine_version)
for plugin_name in required_plugins:
if plugin_name not in uploaded_plugins:
missing_list.append(plugin_name)
return missing_list
[docs]
def get_marketplace_plugins_for_location(project, location):
"""
Gets a list of all the non-built-in plugins needed for a location
Args:
project (str): Project the location belongs to
location (str): Location to check for plugins
Returns:
marketplace_plugin_list (list): List of plugins that are not built-in
Raises:
KeyError: If the uproject file is not in the expected format
"""
marketplace_plugin_list = []
exception_list = get_plugin_exceptions()
uproject_file = get_uproject_path(project, world=location)
if uproject_file:
try:
plugin_list = load_json(uproject_file)["Plugins"]
for plugin in plugin_list:
if plugin.get("MarketplaceURL") or plugin["Name"] in exception_list:
plugin_name = plugin["Name"]
marketplace_plugin_list.append(plugin_name)
return marketplace_plugin_list
except KeyError:
logging.error(
f"Unexpected Format For Location Uproject File\nLocation: {location}\nFile: {uproject_file}"
)
return False
return marketplace_plugin_list
[docs]
def get_installed_marketplace_plugins():
"""
Gets a list of all the marketplace plugins installed to unreal engine
Returns:
plugin_list (list): List of all the marketplace plugins installed to unreal engine
"""
plugin_list = []
plugins_dir = get_unreal_marketplace_plugins_dir()
if plugins_dir:
if not os.path.exists(plugins_dir):
os.mkdir(plugins_dir)
for plugin in os.listdir(plugins_dir):
plugin_list.append(plugin)
return plugin_list
[docs]
def get_plugin_zip_path(plugin_name):
perforce_root = get_perforce_root()
plugin_download_path = os.path.join(os.path.dirname(perforce_root), "UnrealPlugins")
if not os.path.exists(plugin_download_path):
os.mkdir(plugin_download_path)
download_path = os.path.join(plugin_download_path, f"{plugin_name}.zip")
return download_path
[docs]
def download_plugin(plugin_name):
"""
Downloads a plugin from Amazon S3 bucket
Args:
plugin_name (str): Name of the plugin to download
Returns:
download_path (str): Path to the downloaded plugin
"""
cc = ProjectConfig.getLastConfig().cloud_config
if "remote_publish" not in cc:
logging.info("No cloud publish is configured") # remote publish is not enabled.
return
engine_version = get_engine_version()
if not engine_version:
return
handle_aws_login()
download_path = get_plugin_zip_path(plugin_name)
bucket = Cognito.get_s3_bucket(cloud_constants["unreal_plugins_s3"])
object_name = f"{engine_version}/{plugin_name}.zip"
logging.info(f"Downloading '{plugin_name}' from S3")
bucket.download_data(object_name=object_name, file_path=download_path)
return download_path
[docs]
def upload_plugin(plugin_name, zip_file_path):
"""
Uploads a plugin to Amazon S3 bucket
Args:
plugin_name (str): Name of the plugin to upload
zip_file_path (str): Path to the plugin zip file
Returns:
None
"""
cc = ProjectConfig.getLastConfig().cloud_config
if "remote_publish" not in cc:
logging.info("No cloud publish is configured") # remote publish is not enabled.
return
engine_version = get_engine_version()
if not engine_version:
return
handle_aws_login()
bucket = Cognito.get_s3_bucket(cloud_constants["unreal_plugins_s3"])
object_name = f"{plugin_name}.zip"
logging.info(f"Uploading '{plugin_name}' to S3")
bucket.upload_data(
prefix=engine_version, object_name=object_name, file_path=zip_file_path
)
[docs]
def get_plugins_in_s3(engine_version=5.1):
"""
Gets a list of all the plugins in the s3 bucket for the engine version specified
Args:
engine_version (float): Unreal Engine Version to check for plugins. Defaults to 5.1
Returns:
clean_plugin_list (list): List of all the plugins in the s3 bucket for the engine version specified
"""
cc = ProjectConfig.getLastConfig().cloud_config
if "remote_publish" not in cc:
logging.info("No cloud publish is configured") # remote publish is not enabled.
return
handle_aws_login()
bucket = Cognito.get_s3_bucket(cloud_constants["unreal_plugins_s3"])
folder_name = f"{engine_version}/"
uploaded_plugins = bucket.list_folder_contents(folder_path=folder_name)
clean_plugin_list = [
plugin_name.replace(".zip", "") for plugin_name in uploaded_plugins
]
return clean_plugin_list
[docs]
def get_plugin_exceptions():
"""Gets list of exception plugins from cgl.plugins.yaml file.
"Exception" plugins are plugins that won't be listed as "marketplace" plugins in the .uplugin file, but need to
be managed with s3
Returns:
exception_list (list): List of all the exception plugins
"""
plugin_file = os.path.join(CFG.get_code_root(), "config", "plugins.yaml")
plugin_config = load_yaml(plugin_file)
exception_list = plugin_config["marketplace_exceptions"]
return exception_list
[docs]
def unzip_plugin(zip_file_path):
"""
Unzips the plugin file to the Unreal Plugins Folder
Args:
zip_file_path (str): Path to the plugin zip file
"""
plugin_directory = get_unreal_marketplace_plugins_dir()
if not os.path.exists(plugin_directory):
os.mkdir(plugin_directory)
if os.path.exists(zip_file_path):
unzip(zipped_file=zip_file_path, destination=plugin_directory)
[docs]
def zip_plugin(plugin_name):
unreal_plugin_folder = get_unreal_marketplace_plugins_dir()
plugin_folder_path = os.path.join(unreal_plugin_folder, plugin_name)
plugin_zip_path = get_plugin_zip_path(plugin_name=plugin_name)
logging.info(f"Zipping '{plugin_name} Plugin' -> {plugin_zip_path}")
zip_path(source_path=plugin_folder_path, zip_file=plugin_zip_path)
return plugin_zip_path
[docs]
def plugin_download_handler(project, location):
"""
Checks for missing plugins, downloads them, and unzips the plugins to Unreal Engine
Args:
project (str): Project the location belongs to
location (str): Location to check for missing plugins
"""
project_cfg = ProjectConfig().getLastConfig().project_config
network_plugins_dir = project_cfg["unreal_info"]["network_plugins_dir"]
unreal_marketplace_dir = get_unreal_marketplace_plugins_dir()
engine_version = get_engine_version()
if not os.path.exists(unreal_marketplace_dir):
os.mkdir(unreal_marketplace_dir)
#
# if check_for_network_drive_connection(
# "\\\\pixit.fllan.premisestudio.com\\data", "Y:"
# ) or check_for_network_drive_connection("\\\\pixit\\data", "Y:"):
# network_plugins_path = rf"{network_plugins_dir}\{engine_version}\Plugins"
# for folder_name in os.listdir(network_plugins_path):
# network_market_plugins_path = os.path.join(
# network_plugins_path, folder_name
# )
# for folder in os.listdir(network_market_plugins_path):
# if not os.path.exists(os.path.join(unreal_marketplace_dir, folder)):
# cgl_copy(
# os.path.join(network_market_plugins_path, folder),
# os.path.join(unreal_marketplace_dir, folder),
# dest_is_folder=True,
# wait_to_finish=True,
# )
missing_plugins = get_missing_local_plugins(project=project, location=location)
if missing_plugins:
engine_version = get_engine_version()
uploaded_plugins = get_plugins_in_s3(engine_version=engine_version)
for plugin in missing_plugins:
if plugin in uploaded_plugins:
zip_file_path = download_plugin(plugin_name=plugin)
unzip_plugin(zip_file_path=zip_file_path)
else:
logging.error(
f"Plugin '{plugin}' Not Found in S3 for Unreal Engine Version '{engine_version}'\nPlease Upload This Plugin if Its Needed"
)
continue
[docs]
def plugin_upload_handler(project, location):
"""
Checks for plugins that a location needs that aren't uploaded to S3
Args:
project (str): Project the location belongs to
location (str): Location to check for missing plugins
"""
missing_install_plugins = []
plugins_to_upload = get_missing_s3_plugins(project=project, location=location)
installed_plugins = get_installed_marketplace_plugins()
if plugins_to_upload:
for plugin in plugins_to_upload:
if plugin in installed_plugins:
plugin_zip_path = zip_plugin(plugin_name=plugin)
upload_plugin(plugin_name=plugin, zip_file_path=plugin_zip_path)
else:
missing_install_plugins.append(plugin)
else:
logging.info("No Plugins Found that Need to be Uploaded")
if missing_install_plugins:
logging.error(
f"The following plugins are not installed to Unreal\nPlugins: {missing_install_plugins}\n"
f"Please visit Unreal Marketplace and install the plugins and click upload button again if "
f"the plugin is needed."
)