mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-25 13:24:54 +01:00
feat(premiere): wip wrapper
This commit is contained in:
Jakub Jezek
parent
70ef1d8ebc
commit
124e61a847
3 changed files with 183 additions and 234 deletions
|
|
@ -1,11 +1,8 @@
|
|||
import logging
|
||||
import os
|
||||
|
||||
import traceback
|
||||
from pype.lib import PypeHook
|
||||
from pypeapp import Logger
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PremierePrelaunch(PypeHook):
|
||||
"""
|
||||
|
|
@ -26,11 +23,16 @@ class PremierePrelaunch(PypeHook):
|
|||
def execute(self, *args, env: dict = None) -> bool:
|
||||
if not env:
|
||||
env = os.environ
|
||||
|
||||
EXTENSIONS_CACHE_PATH = env.get("EXTENSIONS_CACHE_PATH", None)
|
||||
self.log.debug(
|
||||
"_ EXTENSIONS_CACHE_PATH: `{}`".format(EXTENSIONS_CACHE_PATH))
|
||||
asset = env["AVALON_ASSET"]
|
||||
task = env["AVALON_TASK"]
|
||||
workdir = env["AVALON_WORKDIR"]
|
||||
project_name = f"{asset}_{task}"
|
||||
|
||||
import importlib
|
||||
import avalon.api
|
||||
import pype.premiere
|
||||
avalon.api.install(pype.premiere)
|
||||
|
|
@ -40,13 +42,14 @@ class PremierePrelaunch(PypeHook):
|
|||
__import__("pyblish")
|
||||
|
||||
except ImportError as e:
|
||||
print traceback.format_exc()
|
||||
print(traceback.format_exc())
|
||||
print("pyblish: Could not load integration: %s " % e)
|
||||
|
||||
else:
|
||||
# Setup integration
|
||||
import pype.premiere.lib
|
||||
pype.premiere.lib.setup()
|
||||
from pype.premiere import lib as prlib
|
||||
importlib.reload(prlib)
|
||||
prlib.setup(env)
|
||||
|
||||
self.log.debug("_ self.signature: `{}`".format(self.signature))
|
||||
self.log.debug("_ asset: `{}`".format(asset))
|
||||
|
|
|
|||
|
|
@ -1,167 +1,73 @@
|
|||
import os
|
||||
import sys
|
||||
import shutil
|
||||
|
||||
from pysync import walktree
|
||||
|
||||
from avalon import api as avalon
|
||||
from pyblish import api as pyblish
|
||||
from app import api as app
|
||||
from .. import api
|
||||
import requests
|
||||
from pypeapp import Logger
|
||||
|
||||
from .pipeline import (
|
||||
install,
|
||||
uninstall,
|
||||
|
||||
from .lib import (
|
||||
setup,
|
||||
reload_pipeline,
|
||||
ls
|
||||
ls,
|
||||
LOAD_PATH,
|
||||
INVENTORY_PATH,
|
||||
CREATE_PATH,
|
||||
PUBLISH_PATH,
|
||||
PLUGINS_DIR
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"install",
|
||||
"uninstall",
|
||||
"setup",
|
||||
"reload_pipeline",
|
||||
"ls"
|
||||
]
|
||||
|
||||
log = api.Logger.getLogger(__name__, "premiere")
|
||||
|
||||
AVALON_CONFIG = os.getenv("AVALON_CONFIG", "pype")
|
||||
EXTENSIONS_PATH_LOCAL = os.getenv("EXTENSIONS_PATH", None)
|
||||
EXTENSIONS_CACHE_PATH = os.getenv("EXTENSIONS_CACHE_PATH", None)
|
||||
EXTENSIONS_PATH_REMOTE = os.path.join(os.path.dirname(__file__), "extensions")
|
||||
PARENT_DIR = os.path.dirname(__file__)
|
||||
PACKAGE_DIR = os.path.dirname(PARENT_DIR)
|
||||
PLUGINS_DIR = os.path.join(PACKAGE_DIR, "plugins")
|
||||
|
||||
_clearing_cache = ["com.pype.rename", "com.pype.avalon"]
|
||||
|
||||
PUBLISH_PATH = os.path.join(
|
||||
PLUGINS_DIR, "premiere", "publish"
|
||||
).replace("\\", "/")
|
||||
|
||||
if os.getenv("PUBLISH_PATH", None):
|
||||
os.environ["PUBLISH_PATH"] = os.pathsep.join(
|
||||
os.environ["PUBLISH_PATH"].split(os.pathsep) +
|
||||
[PUBLISH_PATH]
|
||||
)
|
||||
else:
|
||||
os.environ["PUBLISH_PATH"] = PUBLISH_PATH
|
||||
|
||||
LOAD_PATH = os.path.join(PLUGINS_DIR, "premiere", "load")
|
||||
CREATE_PATH = os.path.join(PLUGINS_DIR, "premiere", "create")
|
||||
INVENTORY_PATH = os.path.join(PLUGINS_DIR, "premiere", "inventory")
|
||||
|
||||
log.debug("_clearing_cache: {}".format(_clearing_cache))
|
||||
|
||||
def clearing_caches_ui():
|
||||
'''Before every start of premiere it will make sure there is not
|
||||
outdated stuff in cep_cache dir'''
|
||||
|
||||
for d in os.listdir(EXTENSIONS_CACHE_PATH):
|
||||
match = [p for p in _clearing_cache
|
||||
if str(p) in d]
|
||||
|
||||
if match:
|
||||
try:
|
||||
path = os.path.normpath(os.path.join(EXTENSIONS_CACHE_PATH, d))
|
||||
log.info("Removing dir: {}".format(path))
|
||||
shutil.rmtree(path, ignore_errors=True)
|
||||
except Exception as e:
|
||||
log.debug("problem: {}".format(e))
|
||||
|
||||
def request_aport(url_path, data={}):
|
||||
try:
|
||||
api.add_tool_to_environment(["aport_0.1"])
|
||||
|
||||
ip = os.getenv("PICO_IP", None)
|
||||
if ip and ip.startswith('http'):
|
||||
ip = ip.replace("http://", "")
|
||||
|
||||
port = int(os.getenv("PICO_PORT", None))
|
||||
|
||||
url = "http://{0}:{1}{2}".format(ip, port, url_path)
|
||||
req = requests.post(url, data=data).text
|
||||
return req
|
||||
|
||||
except Exception as e:
|
||||
api.message(title="Premiere Aport Server",
|
||||
message="Before you can run Premiere, start Aport Server. \n Error: {}".format(
|
||||
e),
|
||||
level="critical")
|
||||
|
||||
|
||||
def extensions_sync():
|
||||
# import time
|
||||
process_pairs = list()
|
||||
# get extensions dir in pype.premiere.extensions
|
||||
# build dir path to premiere cep extensions
|
||||
for name in os.listdir(EXTENSIONS_PATH_REMOTE):
|
||||
print(name)
|
||||
src = os.path.join(EXTENSIONS_PATH_REMOTE, name)
|
||||
dst = os.path.join(EXTENSIONS_PATH_LOCAL, name)
|
||||
process_pairs.append((name, src, dst))
|
||||
|
||||
# synchronize all extensions
|
||||
for name, src, dst in process_pairs:
|
||||
if not os.path.exists(dst):
|
||||
os.makedirs(dst, mode=0o777)
|
||||
walktree(source=src, target=dst, options_input=["y", ">"])
|
||||
log.info("Extension {0} from `{1}` coppied to `{2}`".format(
|
||||
name, src, dst
|
||||
))
|
||||
# time.sleep(10)
|
||||
return
|
||||
log = Logger().get_logger(__name__, "premiere")
|
||||
|
||||
|
||||
def install():
|
||||
"""Install Premiere-specific functionality of avalon-core.
|
||||
|
||||
log.info("Registering Premiera plug-ins..")
|
||||
reg_paths = request_aport("/api/register_plugin_path",
|
||||
{"publish_path": PUBLISH_PATH})
|
||||
This is where you install menus and register families, data
|
||||
and loaders into Premiere.
|
||||
|
||||
It is called automatically when installing via `api.install(premiere)`.
|
||||
|
||||
See the Maya equivalent for inspiration on how to implement this.
|
||||
|
||||
"""
|
||||
|
||||
# Disable all families except for the ones we explicitly want to see
|
||||
family_states = [
|
||||
"imagesequence",
|
||||
"mov"
|
||||
|
||||
]
|
||||
avalon.data["familiesStateDefault"] = False
|
||||
avalon.data["familiesStateToggled"] = family_states
|
||||
|
||||
# load data from templates
|
||||
api.load_data_from_templates()
|
||||
log.info("pype.premiere installed")
|
||||
|
||||
# remove cep_cache from user temp dir
|
||||
clearing_caches_ui()
|
||||
pyblish.register_host("premiere")
|
||||
pyblish.register_plugin_path(PUBLISH_PATH)
|
||||
log.info("Registering Premiera plug-ins..")
|
||||
|
||||
# synchronize extensions
|
||||
extensions_sync()
|
||||
message = "The Pype extension has been installed. " \
|
||||
"\nThe following publishing paths has been registered: " \
|
||||
"\n\n{}".format(
|
||||
reg_paths)
|
||||
|
||||
api.message(title="pyblish_paths", message=message, level="info")
|
||||
|
||||
# launching premiere
|
||||
exe = r"C:\Program Files\Adobe\Adobe Premiere Pro CC 2019\Adobe Premiere Pro.exe".replace(
|
||||
"\\", "/")
|
||||
|
||||
log.info("____path exists: {}".format(os.path.exists(exe)))
|
||||
|
||||
app.forward(args=[exe],
|
||||
silent=False,
|
||||
cwd=os.getcwd(),
|
||||
env=dict(os.environ),
|
||||
shell=None)
|
||||
avalon.register_plugin_path(avalon.Loader, LOAD_PATH)
|
||||
avalon.register_plugin_path(avalon.Creator, CREATE_PATH)
|
||||
|
||||
|
||||
def uninstall():
|
||||
log.info("Deregistering Premiera plug-ins..")
|
||||
"""Uninstall all tha was installed
|
||||
|
||||
This is where you undo everything that was done in `install()`.
|
||||
That means, removing menus, deregistering families and data
|
||||
and everything. It should be as though `install()` was never run,
|
||||
because odds are calling this function means the user is interested
|
||||
in re-installing shortly afterwards. If, for example, he has been
|
||||
modifying the menu or registered families.
|
||||
|
||||
"""
|
||||
pyblish.deregister_host("premiere")
|
||||
pyblish.deregister_plugin_path(PUBLISH_PATH)
|
||||
log.info("Deregistering Premiera plug-ins..")
|
||||
|
||||
avalon.deregister_plugin_path(avalon.Loader, LOAD_PATH)
|
||||
avalon.deregister_plugin_path(avalon.Creator, CREATE_PATH)
|
||||
|
||||
# reset data from templates
|
||||
api.reset_data_from_templates()
|
||||
|
|
|
|||
|
|
@ -1,14 +1,49 @@
|
|||
import os
|
||||
import importlib
|
||||
from pyblish import api as pyblish
|
||||
import sys
|
||||
import shutil
|
||||
import json
|
||||
from pysync import walktree
|
||||
import requests
|
||||
|
||||
from avalon import api
|
||||
import logging
|
||||
from pype.widgets.message_window import message
|
||||
from pypeapp import Logger
|
||||
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
log = Logger().get_logger(__name__, "premiere")
|
||||
|
||||
self = sys.modules[__name__]
|
||||
self._has_been_setup = False
|
||||
self._registered_gui = None
|
||||
|
||||
AVALON_CONFIG = os.environ["AVALON_CONFIG"]
|
||||
|
||||
PARENT_DIR = os.path.dirname(__file__)
|
||||
PACKAGE_DIR = os.path.dirname(PARENT_DIR)
|
||||
PLUGINS_DIR = os.path.join(PACKAGE_DIR, "plugins")
|
||||
|
||||
self.EXTENSIONS_PATH_REMOTE = os.path.join(PARENT_DIR, "extensions")
|
||||
self.EXTENSIONS_PATH_LOCAL = None
|
||||
self.EXTENSIONS_CACHE_PATH = None
|
||||
|
||||
self.LOAD_PATH = os.path.join(PLUGINS_DIR, "premiere", "load")
|
||||
self.CREATE_PATH = os.path.join(PLUGINS_DIR, "premiere", "create")
|
||||
self.INVENTORY_PATH = os.path.join(PLUGINS_DIR, "premiere", "inventory")
|
||||
|
||||
self.PUBLISH_PATH = os.path.join(
|
||||
PLUGINS_DIR, "premiere", "publish"
|
||||
).replace("\\", "/")
|
||||
|
||||
if os.getenv("PUBLISH_PATH", None):
|
||||
os.environ["PUBLISH_PATH"] = os.pathsep.join(
|
||||
os.environ["PUBLISH_PATH"].split(os.pathsep) +
|
||||
[self.PUBLISH_PATH]
|
||||
)
|
||||
else:
|
||||
os.environ["PUBLISH_PATH"] = self.PUBLISH_PATH
|
||||
|
||||
_clearing_cache = ["com.pype.rename", "com.pype.avalon"]
|
||||
|
||||
|
||||
def ls():
|
||||
pass
|
||||
|
|
@ -31,107 +66,112 @@ def reload_pipeline():
|
|||
"avalon.api",
|
||||
"avalon.tools",
|
||||
|
||||
"avalon.tools.loader.app",
|
||||
"avalon.tools.creator.app",
|
||||
"avalon.tools.manager.app",
|
||||
|
||||
"avalon.premiere",
|
||||
"avalon.premiere.pipeline",
|
||||
"{}".format(AVALON_CONFIG)
|
||||
"{}".format(AVALON_CONFIG),
|
||||
"{}.premiere".format(AVALON_CONFIG),
|
||||
"{}.premiere.lib".format(AVALON_CONFIG)
|
||||
):
|
||||
log.info("Reloading module: {}...".format(module))
|
||||
module = importlib.import_module(module)
|
||||
reload(module)
|
||||
try:
|
||||
module = importlib.import_module(module)
|
||||
reload(module)
|
||||
except Exception as e:
|
||||
log.warning("Cannot reload module: {}".format(e))
|
||||
importlib.reload(module)
|
||||
|
||||
import avalon.premiere
|
||||
api.install(avalon.premiere)
|
||||
|
||||
|
||||
def install(config):
|
||||
"""Install Premiere-specific functionality of avalon-core.
|
||||
|
||||
This is where you install menus and register families, data
|
||||
and loaders into Premiere.
|
||||
|
||||
It is called automatically when installing via `api.install(premiere)`.
|
||||
|
||||
See the Maya equivalent for inspiration on how to implement this.
|
||||
|
||||
def setup(env=None):
|
||||
""" Running wrapper
|
||||
"""
|
||||
if not env:
|
||||
env = os.environ
|
||||
|
||||
pyblish.register_host("premiere")
|
||||
# Trigger install on the config's "premiere" package
|
||||
config = find_host_config(config)
|
||||
self.EXTENSIONS_PATH_LOCAL = env["EXTENSIONS_PATH"]
|
||||
self.EXTENSIONS_CACHE_PATH = env["EXTENSIONS_CACHE_PATH"]
|
||||
|
||||
if hasattr(config, "install"):
|
||||
config.install()
|
||||
log.info("Registering Premiera plug-ins..")
|
||||
if not test_rest_api_server():
|
||||
return
|
||||
|
||||
log.info("config.premiere installed")
|
||||
# remove cep_cache from user temp dir
|
||||
clearing_caches_ui()
|
||||
|
||||
# synchronize extensions
|
||||
extensions_sync()
|
||||
|
||||
log.info("Premiere Pype wrapper has been installed")
|
||||
|
||||
|
||||
def find_host_config(config):
|
||||
def extensions_sync():
|
||||
# import time
|
||||
process_pairs = list()
|
||||
# get extensions dir in pype.premiere.extensions
|
||||
# build dir path to premiere cep extensions
|
||||
|
||||
for name in os.listdir(self.EXTENSIONS_PATH_REMOTE):
|
||||
log.debug("> name: {}".format(name))
|
||||
src = os.path.join(self.EXTENSIONS_PATH_REMOTE, name)
|
||||
dst = os.path.join(self.EXTENSIONS_PATH_LOCAL, name)
|
||||
process_pairs.append((name, src, dst))
|
||||
|
||||
# synchronize all extensions
|
||||
for name, src, dst in process_pairs:
|
||||
if not os.path.exists(dst):
|
||||
os.makedirs(dst, mode=0o777)
|
||||
walktree(source=src, target=dst, options_input=["y", ">"])
|
||||
log.info("Extension {0} from `{1}` coppied to `{2}`".format(
|
||||
name, src, dst
|
||||
))
|
||||
# time.sleep(10)
|
||||
return
|
||||
|
||||
|
||||
def clearing_caches_ui():
|
||||
'''Before every start of premiere it will make sure there is not
|
||||
outdated stuff in cep_cache dir'''
|
||||
|
||||
if not os.path.isdir(self.EXTENSIONS_CACHE_PATH):
|
||||
os.makedirs(self.EXTENSIONS_CACHE_PATH, mode=0o777)
|
||||
log.info("Created dir: {}".format(self.EXTENSIONS_CACHE_PATH))
|
||||
|
||||
for d in os.listdir(self.EXTENSIONS_CACHE_PATH):
|
||||
match = [p for p in _clearing_cache
|
||||
if str(p) in d]
|
||||
|
||||
if match:
|
||||
try:
|
||||
path = os.path.normpath(
|
||||
os.path.join(self.EXTENSIONS_CACHE_PATH, d))
|
||||
log.info("Removing dir: {}".format(path))
|
||||
shutil.rmtree(path, ignore_errors=True)
|
||||
except Exception as e:
|
||||
log.debug("problem: {}".format(e))
|
||||
|
||||
|
||||
def test_rest_api_server():
|
||||
from pprint import pformat
|
||||
rest_url = os.getenv("PYPE_REST_API_URL")
|
||||
project_name = "{AVALON_PROJECT}".format(**dict(os.environ))
|
||||
URL = "/".join((rest_url,
|
||||
"avalon/projects",
|
||||
project_name))
|
||||
log.debug("__ URL: {}".format(URL))
|
||||
try:
|
||||
config = importlib.import_module(config.__name__ + ".premiere")
|
||||
except ImportError as exc:
|
||||
if str(exc) != "No module name {}".format(
|
||||
config.__name__ + ".premiere"):
|
||||
raise
|
||||
config = None
|
||||
req = requests.get(URL, data={}).text
|
||||
req_json = json.loads(req)
|
||||
# log.debug("_ req_json: {}".format(pformat(req_json)))
|
||||
log.debug("__ projectName: {}".format(req_json["data"]["name"]))
|
||||
assert req_json["data"]["name"] == project_name, (
|
||||
"Project data from Rest API server not correct")
|
||||
return True
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def uninstall(config):
|
||||
"""Uninstall all tha was installed
|
||||
|
||||
This is where you undo everything that was done in `install()`.
|
||||
That means, removing menus, deregistering families and data
|
||||
and everything. It should be as though `install()` was never run,
|
||||
because odds are calling this function means the user is interested
|
||||
in re-installing shortly afterwards. If, for example, he has been
|
||||
modifying the menu or registered families.
|
||||
|
||||
"""
|
||||
config = find_host_config(config)
|
||||
if hasattr(config, "uninstall"):
|
||||
config.uninstall()
|
||||
|
||||
pyblish.deregister_host("premiere")
|
||||
|
||||
|
||||
def get_anatomy(**kwarg):
|
||||
return pype.Anatomy
|
||||
|
||||
|
||||
def get_dataflow(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "premiere")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("premiera.lib.get_dataflow():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
pr_dataflow = getattr(pype.Dataflow, str(host), None)
|
||||
pr_dataflow_node = getattr(pr_dataflow.nodes, str(cls), None)
|
||||
if preset:
|
||||
pr_dataflow_node = getattr(pr_dataflow_node, str(preset), None)
|
||||
|
||||
log.info("Dataflow: {}".format(pr_dataflow_node))
|
||||
return pr_dataflow_node
|
||||
|
||||
|
||||
def get_colorspace(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "premiere")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("premiera.templates.get_colorspace():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
pr_colorspace = getattr(pype.Colorspace, str(host), None)
|
||||
pr_colorspace_node = getattr(pr_colorspace, str(cls), None)
|
||||
if preset:
|
||||
pr_colorspace_node = getattr(pr_colorspace_node, str(preset), None)
|
||||
|
||||
log.info("Colorspace: {}".format(pr_colorspace_node))
|
||||
return pr_colorspace_node
|
||||
except Exception as e:
|
||||
message(title="Pype Rest API static server is not running ",
|
||||
message=("Before you can run Premiere, make sure "
|
||||
"the system Tray Pype icon is running and "
|
||||
"submenu `service` with name `Rest API` is "
|
||||
"with green icon."
|
||||
"\n Error: {}".format(e)),
|
||||
level="critical")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue