mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-24 12:54:40 +01:00
adding context related functions to pype/template.py and other small changes, fixing typo on get_hierarchy(), rename fill_avalon_workdir to set_*, rename make_wordir_template to get_*,
This commit is contained in:
parent
eaa22dc082
commit
61bbfebb1e
20 changed files with 475 additions and 178 deletions
26
pype/api.py
26
pype/api.py
|
|
@ -29,15 +29,19 @@ from .templates import (
|
|||
reset_data_from_templates,
|
||||
get_project_name,
|
||||
get_project_code,
|
||||
get_hiearchy,
|
||||
get_hierarchy,
|
||||
get_asset,
|
||||
get_task,
|
||||
fill_avalon_workdir,
|
||||
set_avalon_workdir,
|
||||
get_version_from_workfile,
|
||||
make_workdir_path
|
||||
get_workdir_template,
|
||||
set_hierarchy,
|
||||
set_project_code
|
||||
)
|
||||
|
||||
from .lib import modified_environ
|
||||
from .lib import modified_environ, add_tool_to_environment
|
||||
|
||||
from .widgets.message_window import message
|
||||
|
||||
__all__ = [
|
||||
# plugin classes
|
||||
|
|
@ -61,18 +65,24 @@ __all__ = [
|
|||
# get contextual data
|
||||
"get_project_name",
|
||||
"get_project_code",
|
||||
"get_hiearchy",
|
||||
"get_hierarchy",
|
||||
"get_asset",
|
||||
"get_task",
|
||||
"fill_avalon_workdir",
|
||||
"set_avalon_workdir",
|
||||
"get_version_from_workfile",
|
||||
"make_workdir_path",
|
||||
"get_workdir_template",
|
||||
"modified_environ",
|
||||
"add_tool_to_environment",
|
||||
"set_hierarchy",
|
||||
"set_project_code",
|
||||
|
||||
# preloaded templates
|
||||
"Anatomy",
|
||||
"Colorspace",
|
||||
"Metadata",
|
||||
"Dataflow"
|
||||
"Dataflow",
|
||||
|
||||
# QtWidgets
|
||||
"message"
|
||||
|
||||
]
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ from app import api as app
|
|||
|
||||
from .. import api
|
||||
|
||||
log = api.Logger.getLogger(__name__, "editorial")
|
||||
log = api.Logger.getLogger(__name__, "aport")
|
||||
|
||||
AVALON_CONFIG = os.getenv("AVALON_CONFIG", "pype")
|
||||
|
||||
|
|
@ -16,8 +16,8 @@ PACKAGE_DIR = os.path.dirname(PARENT_DIR)
|
|||
PLUGINS_DIR = os.path.join(PACKAGE_DIR, "plugins")
|
||||
|
||||
PUBLISH_PATH = os.path.join(
|
||||
PLUGINS_DIR, "editorial", "publish"
|
||||
)
|
||||
PLUGINS_DIR, "aport", "publish"
|
||||
).replace("\\", "/")
|
||||
|
||||
if os.getenv("PUBLISH_PATH", None):
|
||||
os.environ["PUBLISH_PATH"] = os.pathsep.join(
|
||||
|
|
@ -27,15 +27,15 @@ if os.getenv("PUBLISH_PATH", None):
|
|||
else:
|
||||
os.environ["PUBLISH_PATH"] = PUBLISH_PATH
|
||||
|
||||
LOAD_PATH = os.path.join(PLUGINS_DIR, "editorial", "load")
|
||||
CREATE_PATH = os.path.join(PLUGINS_DIR, "editorial", "create")
|
||||
INVENTORY_PATH = os.path.join(PLUGINS_DIR, "editorial", "inventory")
|
||||
LOAD_PATH = os.path.join(PLUGINS_DIR, "aport", "load")
|
||||
CREATE_PATH = os.path.join(PLUGINS_DIR, "aport", "create")
|
||||
INVENTORY_PATH = os.path.join(PLUGINS_DIR, "aport", "inventory")
|
||||
|
||||
|
||||
def install():
|
||||
api.fill_avalon_workdir()
|
||||
api.set_avalon_workdir()
|
||||
|
||||
log.info("Registering Editorial plug-ins..")
|
||||
log.info("Registering Aport plug-ins..")
|
||||
pyblish.register_plugin_path(PUBLISH_PATH)
|
||||
avalon.register_plugin_path(avalon.Loader, LOAD_PATH)
|
||||
avalon.register_plugin_path(avalon.Creator, CREATE_PATH)
|
||||
|
|
@ -58,7 +58,7 @@ def install():
|
|||
|
||||
|
||||
def uninstall():
|
||||
log.info("Deregistering Editorial plug-ins..")
|
||||
log.info("Deregistering Aport plug-ins..")
|
||||
pyblish.deregister_plugin_path(PUBLISH_PATH)
|
||||
avalon.deregister_plugin_path(avalon.Loader, LOAD_PATH)
|
||||
avalon.deregister_plugin_path(avalon.Creator, CREATE_PATH)
|
||||
|
|
@ -71,7 +71,7 @@ def pico_server_launch():
|
|||
try:
|
||||
args = [sys.executable, "-m", "pico.server", "pipeline"]
|
||||
|
||||
returncode = app.forward(
|
||||
app.forward(
|
||||
args,
|
||||
cwd=os.path.dirname(__file__)
|
||||
)
|
||||
|
|
@ -10,8 +10,8 @@
|
|||
<body>
|
||||
<p id="message"></p>
|
||||
<script>
|
||||
var example = pico.importModule('example')
|
||||
example.hello("Fergal").then(function(response){
|
||||
var pipeline = pico.importModule('pipeline')
|
||||
pipeline.context("Fergal").then(function(response){
|
||||
document.getElementById('message').innerHTML = response;
|
||||
});
|
||||
</script>
|
||||
105
pype/aport/pipeline.py
Normal file
105
pype/aport/pipeline.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
import os
|
||||
import sys
|
||||
import pico
|
||||
# from pico.decorators import request_args, prehandle
|
||||
from pico import PicoApp
|
||||
from pico import client
|
||||
|
||||
from avalon import api as avalon
|
||||
from avalon import io
|
||||
|
||||
import pyblish.api as pyblish
|
||||
|
||||
from app.api import forward
|
||||
from pype import api as pype
|
||||
|
||||
# remove all Handlers created by pico
|
||||
for name, handler in [(handler.get_name(), handler)
|
||||
for handler in pype.Logger.logging.root.handlers[:]]:
|
||||
if "pype" not in str(name).lower():
|
||||
pype.Logger.logging.root.removeHandler(handler)
|
||||
|
||||
log = pype.Logger.getLogger(__name__, "aport")
|
||||
|
||||
|
||||
SESSION = avalon.session
|
||||
if not SESSION:
|
||||
io.install()
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def publish(json_data_path):
|
||||
log.warning("avalon.session is: \n{}".format(SESSION))
|
||||
# load json_data_path; add context into data; damp
|
||||
# create empty temp/json_data_get
|
||||
# run standalone pyblish
|
||||
pype_start = os.path.join(os.getenv('PYPE_SETUP_ROOT'),
|
||||
"app", "pype-start.py")
|
||||
|
||||
args = [pype_start, "--publish",
|
||||
"-pp", os.environ["PUBLISH_PATH"],
|
||||
"-d", "json_context_data_path", json_data_path
|
||||
]
|
||||
|
||||
log.info(args)
|
||||
|
||||
# start standalone pyblish qml
|
||||
forward([
|
||||
sys.executable, "-u"
|
||||
] + args,
|
||||
cwd=os.getenv('PYPE_SETUP_ROOT')
|
||||
)
|
||||
|
||||
return {"json_back": "this/json/file"}
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def context(project, asset, task, app):
|
||||
# http://localhost:4242/pipeline/context?project=this&asset=shot01&task=comp
|
||||
|
||||
os.environ["AVALON_PROJECT"] = project
|
||||
|
||||
avalon.update_current_task(task, asset, app)
|
||||
|
||||
project_code = pype.get_project_code()
|
||||
pype.set_project_code(project_code)
|
||||
hierarchy = pype.get_hierarchy()
|
||||
pype.set_hierarchy(hierarchy)
|
||||
SESSION.update({"AVALON_HIERARCHY": hierarchy,
|
||||
"AVALON_PROJECTCODE": project_code,
|
||||
"current_dir": os.getcwd()
|
||||
})
|
||||
|
||||
return SESSION
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def deregister_plugin_path():
|
||||
if os.getenv("PUBLISH_PATH", None):
|
||||
aport_plugin_path = [p.replace("\\", "/") for p in os.environ["PUBLISH_PATH"].split(
|
||||
os.pathsep) if "aport" in p][0]
|
||||
os.environ["PUBLISH_PATH"] = aport_plugin_path
|
||||
else:
|
||||
log.warning("deregister_plugin_path(): No PUBLISH_PATH is registred")
|
||||
|
||||
return "Publish path deregistered"
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def register_plugin_path(publish_path):
|
||||
deregister_plugin_path()
|
||||
if os.getenv("PUBLISH_PATH", None):
|
||||
os.environ["PUBLISH_PATH"] = os.pathsep.join(
|
||||
os.environ["PUBLISH_PATH"].split(os.pathsep) +
|
||||
[publish_path.replace("\\", "/")]
|
||||
)
|
||||
else:
|
||||
os.environ["PUBLISH_PATH"] = publish_path
|
||||
log.warning(os.environ["PUBLISH_PATH"].split(os.pathsep))
|
||||
return "Publish registered paths: {}".format(
|
||||
os.environ["PUBLISH_PATH"].split(os.pathsep)
|
||||
)
|
||||
|
||||
|
||||
app = PicoApp()
|
||||
app.register_module(__name__)
|
||||
41
pype/aport/templates.py
Normal file
41
pype/aport/templates.py
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
from pype import api as pype
|
||||
|
||||
log = pype.Logger.getLogger(__name__, "aport")
|
||||
|
||||
|
||||
def get_anatomy(**kwarg):
|
||||
return pype.Anatomy
|
||||
|
||||
|
||||
def get_dataflow(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "aport")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("aport.templates.get_dataflow():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
aport_dataflow = getattr(pype.Dataflow, str(host), None)
|
||||
aport_dataflow_node = getattr(aport_dataflow.nodes, str(cls), None)
|
||||
if preset:
|
||||
aport_dataflow_node = getattr(aport_dataflow_node, str(preset), None)
|
||||
|
||||
log.info("Dataflow: {}".format(aport_dataflow_node))
|
||||
return aport_dataflow_node
|
||||
|
||||
|
||||
def get_colorspace(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "aport")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("aport.templates.get_colorspace():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
aport_colorspace = getattr(pype.Colorspace, str(host), None)
|
||||
aport_colorspace_node = getattr(aport_colorspace, str(cls), None)
|
||||
if preset:
|
||||
aport_colorspace_node = getattr(aport_colorspace_node, str(preset), None)
|
||||
|
||||
log.info("Colorspace: {}".format(aport_colorspace_node))
|
||||
return aport_colorspace_node
|
||||
|
|
@ -1,44 +0,0 @@
|
|||
import os
|
||||
import sys
|
||||
import pico
|
||||
from pico import PicoApp
|
||||
|
||||
from app.api import forward
|
||||
from pype import api as pype
|
||||
|
||||
# remove all Handlers created by pico
|
||||
for name, handler in [(handler.get_name(), handler)
|
||||
for handler in pype.Logger.logging.root.handlers[:]]:
|
||||
if "pype" not in str(name).lower():
|
||||
pype.Logger.logging.root.removeHandler(handler)
|
||||
|
||||
log = pype.Logger.getLogger(__name__, "editorial")
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def publish(json_data_path):
|
||||
# load json_data_path; add context into data; damp
|
||||
# create empty temp/json_data_get
|
||||
# run standalone pyblish
|
||||
pype_start = os.path.join(os.getenv('PYPE_SETUP_ROOT'),
|
||||
"app", "pype-start.py")
|
||||
|
||||
args = [pype_start, "--publish",
|
||||
"-pp", os.environ["PUBLISH_PATH"],
|
||||
"-d", "json_context_data_path", json_data_path
|
||||
]
|
||||
|
||||
log.info(args)
|
||||
|
||||
# start standalone pyblish qml
|
||||
forward([
|
||||
sys.executable, "-u"
|
||||
] + args,
|
||||
cwd=os.getenv('PYPE_SETUP_ROOT')
|
||||
)
|
||||
|
||||
return {"json_back": "this/json/file"}
|
||||
|
||||
|
||||
app = PicoApp()
|
||||
app.register_module(__name__)
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
from pype import api as pype
|
||||
|
||||
log = pype.Logger.getLogger(__name__, "editorial")
|
||||
|
||||
|
||||
def get_anatomy(**kwarg):
|
||||
return pype.Anatomy
|
||||
|
||||
|
||||
def get_dataflow(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "editorial")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("premiera.templates.get_dataflow():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
edtr_dataflow = getattr(pype.Dataflow, str(host), None)
|
||||
edtr_dataflow_node = getattr(edtr_dataflow.nodes, str(cls), None)
|
||||
if preset:
|
||||
edtr_dataflow_node = getattr(edtr_dataflow_node, str(preset), None)
|
||||
|
||||
log.info("Dataflow: {}".format(edtr_dataflow_node))
|
||||
return edtr_dataflow_node
|
||||
|
||||
|
||||
def get_colorspace(**kwarg):
|
||||
log.info(kwarg)
|
||||
host = kwarg.get("host", "editorial")
|
||||
cls = kwarg.get("class", None)
|
||||
preset = kwarg.get("preset", None)
|
||||
assert any([host, cls]), log.error("editorial.templates.get_colorspace():"
|
||||
"Missing mandatory kwargs `host`, `cls`")
|
||||
|
||||
edtr_colorspace = getattr(pype.Colorspace, str(host), None)
|
||||
edtr_colorspace_node = getattr(edtr_colorspace, str(cls), None)
|
||||
if preset:
|
||||
edtr_colorspace_node = getattr(edtr_colorspace_node, str(preset), None)
|
||||
|
||||
log.info("Colorspace: {}".format(edtr_colorspace_node))
|
||||
return edtr_colorspace_node
|
||||
21
pype/lib.py
21
pype/lib.py
|
|
@ -15,6 +15,24 @@ import avalon
|
|||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def add_tool_to_environment(tools):
|
||||
"""
|
||||
It is adding dynamic environment to os environment.
|
||||
|
||||
Args:
|
||||
tool (list, tuple): list of tools, name should corespond to json/toml
|
||||
|
||||
Returns:
|
||||
os.environ[KEY]: adding to os.environ
|
||||
"""
|
||||
|
||||
import acre
|
||||
tools_env = acre.get_tools(tools)
|
||||
env = acre.compute(tools_env)
|
||||
env = acre.merge(env, current_env=dict(os.environ))
|
||||
os.environ.update(env)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def modified_environ(*remove, **update):
|
||||
"""
|
||||
|
|
@ -383,7 +401,8 @@ def get_avalon_project_template_schema():
|
|||
def get_avalon_project_template():
|
||||
from app.api import Templates
|
||||
|
||||
"""Get avalon template
|
||||
"""
|
||||
Get avalon template
|
||||
|
||||
Returns:
|
||||
dictionary with templates
|
||||
|
|
|
|||
|
|
@ -99,7 +99,7 @@ def reload_config():
|
|||
|
||||
def install():
|
||||
|
||||
api.fill_avalon_workdir()
|
||||
api.set_avalon_workdir()
|
||||
reload_config()
|
||||
|
||||
log.info("Registering Nuke plug-ins..")
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ def format_anatomy(data):
|
|||
padding = anatomy.render.padding
|
||||
|
||||
data.update({
|
||||
"hierarchy": pype.get_hiearchy(),
|
||||
"hierarchy": pype.get_hierarchy(),
|
||||
"frame": "#"*padding,
|
||||
"VERSION": pype.get_version_from_workfile(file)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
import pyblish.api
|
||||
from avalon import api as avalon
|
||||
from avalon import io
|
||||
|
||||
|
||||
class CollectContextDataEditorial(pyblish.api.ContextPlugin):
|
||||
|
|
@ -10,3 +12,4 @@ class CollectContextDataEditorial(pyblish.api.ContextPlugin):
|
|||
def process(self, context):
|
||||
data_path = context.data['json_context_data_path']
|
||||
self.log.info("Context is: {}".format(data_path))
|
||||
self.log.warning("avalon.session is: {}".format(avalon.session))
|
||||
|
|
@ -9,13 +9,13 @@ import pype.api as pype
|
|||
|
||||
from pype.api import Logger
|
||||
|
||||
log = Logger.getLogger(__name__, "editorial")
|
||||
log = Logger.getLogger(__name__, "aport")
|
||||
|
||||
|
||||
class Editorial(api.Action):
|
||||
class Aport(api.Action):
|
||||
|
||||
name = "editorial"
|
||||
label = "Pype Editorial Server"
|
||||
name = "aport"
|
||||
label = "Aport - Avalon's Server"
|
||||
icon = "retweet"
|
||||
order = 996
|
||||
|
||||
|
|
@ -38,6 +38,7 @@ class Editorial(api.Action):
|
|||
|
||||
with pype.modified_environ(**session):
|
||||
# Get executable by name
|
||||
print(self.name)
|
||||
app = lib.get_application(self.name)
|
||||
executable = lib.which(app["executable"])
|
||||
|
||||
|
|
@ -49,10 +50,7 @@ class Editorial(api.Action):
|
|||
env = acre.merge(env, current_env=dict(os.environ))
|
||||
|
||||
if not env.get('AVALON_WORKDIR', None):
|
||||
pype.load_data_from_templates()
|
||||
os.environ["AVALON_WORKDIR"] = pype.make_workdir_path(
|
||||
pype.Anatomy)
|
||||
pype.reset_data_from_templates()
|
||||
os.environ["AVALON_WORKDIR"] = pype.get_workdir_template()
|
||||
|
||||
env.update(dict(os.environ))
|
||||
|
||||
|
|
@ -45,7 +45,7 @@ class PremierePro(api.Action):
|
|||
|
||||
if not env.get('AVALON_WORKDIR', None):
|
||||
pype.load_data_from_templates()
|
||||
os.environ["AVALON_WORKDIR"] = pype.make_workdir_path(
|
||||
os.environ["AVALON_WORKDIR"] = pype.get_workdir_template(
|
||||
pype.Anatomy)
|
||||
pype.reset_data_from_templates()
|
||||
|
||||
|
|
|
|||
|
|
@ -6,9 +6,12 @@ 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
|
||||
from pprint import pprint
|
||||
from .. import api
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
log = api.Logger.getLogger(__name__, "premiere")
|
||||
|
||||
AVALON_CONFIG = os.getenv("AVALON_CONFIG", "pype")
|
||||
|
|
@ -20,13 +23,29 @@ PLUGINS_DIR = os.path.join(PACKAGE_DIR, "plugins")
|
|||
|
||||
PUBLISH_PATH = os.path.join(
|
||||
PLUGINS_DIR, "premiera", "publish"
|
||||
)
|
||||
).replace("\\", "/")
|
||||
|
||||
LOAD_PATH = os.path.join(PLUGINS_DIR, "premiera", "load")
|
||||
CREATE_PATH = os.path.join(PLUGINS_DIR, "premiera", "create")
|
||||
INVENTORY_PATH = os.path.join(PLUGINS_DIR, "premiera", "inventory")
|
||||
|
||||
|
||||
def request_aport(url_path, data={}):
|
||||
try:
|
||||
api.add_tool_to_environment(["aport"])
|
||||
ip = os.getenv("PICO_IP", None)
|
||||
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()
|
||||
|
|
@ -51,10 +70,13 @@ def extensions_sync():
|
|||
|
||||
|
||||
def install():
|
||||
api.fill_avalon_workdir()
|
||||
|
||||
api.set_avalon_workdir()
|
||||
log.info("Registering Premiera plug-ins..")
|
||||
pyblish.register_plugin_path(PUBLISH_PATH)
|
||||
reg_paths = request_aport("/pipeline/register_plugin_path",
|
||||
{"publish_path": PUBLISH_PATH})
|
||||
api.message(title="pyblish_paths", message=str(reg_paths), level="info")
|
||||
|
||||
avalon.register_plugin_path(avalon.Loader, LOAD_PATH)
|
||||
avalon.register_plugin_path(avalon.Creator, CREATE_PATH)
|
||||
avalon.register_plugin_path(avalon.InventoryAction, INVENTORY_PATH)
|
||||
|
|
|
|||
|
|
@ -1,5 +0,0 @@
|
|||
import pico.client
|
||||
|
||||
|
||||
example = pico.client.load('http://localhost:4242/example')
|
||||
print(example.hello('You bastard'))
|
||||
|
|
@ -1,18 +0,0 @@
|
|||
import pico
|
||||
from pico import PicoApp
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def hello(who):
|
||||
s = "hello %s!" % who
|
||||
return s
|
||||
|
||||
|
||||
@pico.expose()
|
||||
def goodbye(who):
|
||||
s = "goodbye %s!" % who
|
||||
return s
|
||||
|
||||
|
||||
app = PicoApp()
|
||||
app.register_module(__name__)
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
pushd %~dp0
|
||||
python -m pico.server example
|
||||
REM python -m pico.server
|
||||
|
|
@ -1,12 +1,27 @@
|
|||
import os
|
||||
import re
|
||||
from avalon import io
|
||||
from avalon import api as avalon
|
||||
|
||||
from app.api import (Templates, Logger, format)
|
||||
log = Logger.getLogger(__name__,
|
||||
os.getenv("AVALON_APP", "pype-config"))
|
||||
|
||||
SESSION = avalon.session
|
||||
if not SESSION:
|
||||
io.install()
|
||||
|
||||
|
||||
def load_data_from_templates():
|
||||
"""
|
||||
Load Templates `contextual` data as singleton object
|
||||
[info](https://en.wikipedia.org/wiki/Singleton_pattern)
|
||||
|
||||
Returns:
|
||||
singleton: adding data to sharable object variable
|
||||
|
||||
"""
|
||||
|
||||
from . import api
|
||||
if not any([
|
||||
api.Dataflow,
|
||||
|
|
@ -26,6 +41,15 @@ def load_data_from_templates():
|
|||
|
||||
|
||||
def reset_data_from_templates():
|
||||
"""
|
||||
Clear Templates `contextual` data from singleton
|
||||
object variable
|
||||
|
||||
Returns:
|
||||
singleton: clearing data to None
|
||||
|
||||
"""
|
||||
|
||||
from . import api
|
||||
api.Dataflow = None
|
||||
api.Anatomy = None
|
||||
|
|
@ -35,10 +59,20 @@ def reset_data_from_templates():
|
|||
|
||||
|
||||
def get_version_from_workfile(file):
|
||||
"""
|
||||
Finds version number in file path string
|
||||
|
||||
Args:
|
||||
file (string): file path
|
||||
|
||||
Returns:
|
||||
v: version number in string ('001')
|
||||
|
||||
"""
|
||||
pattern = re.compile(r"_v([0-9]*)")
|
||||
try:
|
||||
v_string = pattern.findall(file)[0]
|
||||
return v_string
|
||||
v = pattern.findall(file)[0]
|
||||
return v
|
||||
except IndexError:
|
||||
log.error("templates:get_version_from_workfile:"
|
||||
"`{}` missing version string."
|
||||
|
|
@ -46,31 +80,93 @@ def get_version_from_workfile(file):
|
|||
|
||||
|
||||
def get_project_code():
|
||||
"""
|
||||
Obtain project code from database
|
||||
|
||||
Returns:
|
||||
string: project code
|
||||
"""
|
||||
|
||||
return io.find_one({"type": "project"})["data"]["code"]
|
||||
|
||||
|
||||
def set_project_code(code):
|
||||
"""
|
||||
Set project code into os.environ
|
||||
|
||||
Args:
|
||||
code (string): project code
|
||||
|
||||
Returns:
|
||||
os.environ[KEY]: project code
|
||||
avalon.sesion[KEY]: project code
|
||||
"""
|
||||
SESSION["AVALON_PROJECTCODE"] = code
|
||||
os.environ["AVALON_PROJECTCODE"] = code
|
||||
|
||||
|
||||
def get_project_name():
|
||||
project_name = os.getenv("AVALON_PROJECT", None)
|
||||
"""
|
||||
Obtain project name from environment variable
|
||||
|
||||
Returns:
|
||||
string: project name
|
||||
|
||||
"""
|
||||
|
||||
project_name = SESSION.get("AVALON_PROJECT", None) \
|
||||
or os.getenv("AVALON_PROJECT", None)
|
||||
assert project_name, log.error("missing `AVALON_PROJECT`"
|
||||
"in environment variables")
|
||||
"in avalon session "
|
||||
"or os.environ!")
|
||||
return project_name
|
||||
|
||||
|
||||
def get_asset():
|
||||
asset = os.getenv("AVALON_ASSET", None)
|
||||
"""
|
||||
Obtain Asset string from session or environment variable
|
||||
|
||||
Returns:
|
||||
string: asset name
|
||||
|
||||
Raises:
|
||||
log: error
|
||||
"""
|
||||
|
||||
asset = SESSION.get("AVALON_ASSET", None) \
|
||||
or os.getenv("AVALON_ASSET", None)
|
||||
assert asset, log.error("missing `AVALON_ASSET`"
|
||||
"in environment variables")
|
||||
"in avalon session "
|
||||
"or os.environ!")
|
||||
return asset
|
||||
|
||||
|
||||
def get_task():
|
||||
task = os.getenv("AVALON_TASK", None)
|
||||
"""
|
||||
Obtain Task string from session or environment variable
|
||||
|
||||
Returns:
|
||||
string: task name
|
||||
|
||||
Raises:
|
||||
log: error
|
||||
"""
|
||||
task = SESSION.get("AVALON_TASK", None) \
|
||||
or os.getenv("AVALON_TASK", None)
|
||||
assert task, log.error("missing `AVALON_TASK`"
|
||||
"in environment variables")
|
||||
"in avalon session "
|
||||
"or os.environ!")
|
||||
return task
|
||||
|
||||
|
||||
def get_hiearchy():
|
||||
def get_hierarchy():
|
||||
"""
|
||||
Obtain asset hierarchy path string from mongo db
|
||||
|
||||
Returns:
|
||||
string: asset hierarchy path
|
||||
|
||||
"""
|
||||
hierarchy = io.find_one({
|
||||
"type": 'asset',
|
||||
"name": get_asset()}
|
||||
|
|
@ -78,38 +174,102 @@ def get_hiearchy():
|
|||
|
||||
if hierarchy:
|
||||
# hierarchy = os.path.sep.join(hierarchy)
|
||||
return os.path.join(*hierarchy)
|
||||
return os.path.join(*hierarchy).replace("\\", "/")
|
||||
|
||||
|
||||
def fill_avalon_workdir():
|
||||
awd = os.getenv("AVALON_WORKDIR", None)
|
||||
assert awd, log.error("missing `AVALON_WORKDIR`"
|
||||
"in environment variables")
|
||||
if "{" not in awd:
|
||||
return
|
||||
def set_hierarchy(hierarchy):
|
||||
"""
|
||||
Updates os.environ and session with asset hierarchy
|
||||
|
||||
Args:
|
||||
hierarchy (string): hierarchy path ("silo/folder/seq")
|
||||
"""
|
||||
SESSION["AVALON_HIERARCHY"] = hierarchy
|
||||
os.environ["AVALON_HIERARCHY"] = hierarchy
|
||||
|
||||
|
||||
def get_context_data(project=None,
|
||||
hierarchy=None,
|
||||
asset=None,
|
||||
task=None):
|
||||
"""
|
||||
Collect all main contextual data
|
||||
|
||||
Args:
|
||||
project (string, optional): project name
|
||||
hierarchy (string, optional): hierarchy path
|
||||
asset (string, optional): asset name
|
||||
task (string, optional): task name
|
||||
|
||||
Returns:
|
||||
dict: contextual data
|
||||
|
||||
"""
|
||||
|
||||
data = {
|
||||
"hierarchy": get_hiearchy(),
|
||||
"task": get_task(),
|
||||
"asset": get_asset(),
|
||||
"project": {"name": get_project_name(),
|
||||
"code": get_project_code()}}
|
||||
"task": task or get_task(),
|
||||
"asset": asset or get_asset(),
|
||||
"project": {"name": project or get_project_name(),
|
||||
"code": get_project_code()},
|
||||
"hierarchy": hierarchy or get_hierarchy(),
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def set_avalon_workdir(project=None,
|
||||
hierarchy=None,
|
||||
asset=None,
|
||||
task=None):
|
||||
"""
|
||||
Updates os.environ and session with filled workdir
|
||||
|
||||
Args:
|
||||
project (string, optional): project name
|
||||
hierarchy (string, optional): hierarchy path
|
||||
asset (string, optional): asset name
|
||||
task (string, optional): task name
|
||||
|
||||
Returns:
|
||||
os.environ[AVALON_WORKDIR]: workdir path
|
||||
avalon.session[AVALON_WORKDIR]: workdir path
|
||||
|
||||
"""
|
||||
awd = SESSION.get("AVALON_WORKDIR", None) \
|
||||
or os.getenv("AVALON_WORKDIR", None)
|
||||
data = get_context_data(project, hierarchy, asset, task)
|
||||
|
||||
if (not awd) or ("{" not in awd):
|
||||
awd = get_workdir_template(data)
|
||||
|
||||
awd_filled = os.path.normpath(format(awd, data))
|
||||
|
||||
SESSION["AVALON_WORKDIR"] = awd_filled
|
||||
os.environ["AVALON_WORKDIR"] = awd_filled
|
||||
log.info("`AVALON_WORKDIR` fixed to: {}".format(awd_filled))
|
||||
|
||||
|
||||
def make_workdir_path(anatomy):
|
||||
try:
|
||||
data = {"project": {"name": get_project_name(),
|
||||
"code": get_project_code()},
|
||||
"task": get_task(),
|
||||
"asset": get_asset(),
|
||||
"hierarchy": os.environ["AVALON_HIERARCHY"]}
|
||||
def get_workdir_template(data=None):
|
||||
"""
|
||||
Obtain workdir templated path from api.Anatomy singleton
|
||||
|
||||
anatomy = anatomy.format(data)
|
||||
Args:
|
||||
data (dict, optional): basic contextual data
|
||||
|
||||
Returns:
|
||||
string: template path
|
||||
"""
|
||||
from . import api
|
||||
|
||||
""" Installs singleton data """
|
||||
load_data_from_templates()
|
||||
|
||||
anatomy = api.Anatomy
|
||||
|
||||
try:
|
||||
anatomy = anatomy.format(data or get_context_data())
|
||||
except Exception as e:
|
||||
log.error("{0} Error in anatomy.format: {1}".format(__name__, e))
|
||||
log.error("{0} Error in "
|
||||
"get_workdir_template(): {1}".format(__name__, e))
|
||||
|
||||
return os.path.join(anatomy.work.root,
|
||||
anatomy.work.folder)
|
||||
|
|
|
|||
50
pype/widgets/message_window.py
Normal file
50
pype/widgets/message_window.py
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
import sys
|
||||
import logging
|
||||
|
||||
from avalon.vendor.Qt.QtWidgets import QApplication, QWidget, QPushButton, QMessageBox
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Window(QWidget):
|
||||
def __init__(self, title, message, level):
|
||||
super().__init__()
|
||||
self.title = title
|
||||
self.message = message
|
||||
self.level = level
|
||||
|
||||
if self.level is "info":
|
||||
self._info()
|
||||
elif self.level is "warning":
|
||||
self._warning()
|
||||
elif self.level is "critical":
|
||||
self._critical()
|
||||
|
||||
def _info(self):
|
||||
self.setWindowTitle(self.title)
|
||||
rc = QMessageBox.information(
|
||||
self, self.title, self.message)
|
||||
if rc:
|
||||
sys.exit(app.exec_())
|
||||
|
||||
def _warning(self):
|
||||
self.setWindowTitle(self.title)
|
||||
rc = QMessageBox.warning(
|
||||
self, self.title, self.message)
|
||||
if rc:
|
||||
sys.exit(app.exec_())
|
||||
|
||||
def _critical(self):
|
||||
self.setWindowTitle(self.title)
|
||||
rc = QMessageBox.critical(
|
||||
self, self.title, self.message)
|
||||
if rc:
|
||||
sys.exit(app.exec_())
|
||||
|
||||
|
||||
def message(title=None, message=None, level="info"):
|
||||
global app
|
||||
app = QApplication(sys.argv)
|
||||
ex = Window(title, message, level)
|
||||
ex.show()
|
||||
sys.exit(app.exec_())
|
||||
Loading…
Add table
Add a link
Reference in a new issue