Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2025-12-24 21:04:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove unused teshost

Browse source
This commit is contained in:
Jakub Trllo 2022-09-01 12:59:14 +02:00
parent a515e58196
commit aaff1525a0
14 changed files with 0 additions and 834 deletions
Download patch file Download diff file Expand all files Collapse all files

16
openpype/hosts/testhost/README.md
View file

@ -1,16 +0,0 @@
# What is `testhost`
Host `testhost` was created to fake running host for testing of publisher.
Does not have any proper launch mechanism at the moment. There is python script `./run_publish.py` which will show publisher window. The script requires to set few variables to run. Execution will register host `testhost`, register global publish plugins and register creator and publish plugins from `./plugins`.
## Data
Created instances and context data are stored into json files inside `./api` folder. Can be easily modified to save them to a different place.
## Plugins
Test host has few plugins to be able test publishing.
### Creators
They are just example plugins using functions from `api` to create/remove/update data. One of them is auto creator which means that is triggered on each reset of create context. Others are manual creators both creating the same family.
### Publishers
Collectors are example plugin to use `get_attribute_defs` to define attributes for specific families or for context. Validators are to test `PublishValidationError`.

0
openpype/hosts/testhost/__init__.py
View file

43
openpype/hosts/testhost/api/__init__.py
View file

@ -1,43 +0,0 @@
import os
import logging
import pyblish.api
from openpype.pipeline import register_creator_plugin_path
from .pipeline import (
ls,
list_instances,
update_instances,
remove_instances,
get_context_data,
update_context_data,
get_context_title
)
HOST_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PLUGINS_DIR = os.path.join(HOST_DIR, "plugins")
PUBLISH_PATH = os.path.join(PLUGINS_DIR, "publish")
CREATE_PATH = os.path.join(PLUGINS_DIR, "create")
log = logging.getLogger(__name__)
def install():
log.info("OpenPype - Installing TestHost integration")
pyblish.api.register_host("testhost")
pyblish.api.register_plugin_path(PUBLISH_PATH)
register_creator_plugin_path(CREATE_PATH)
__all__ = (
"ls",
"list_instances",
"update_instances",
"remove_instances",
"get_context_data",
"update_context_data",
"get_context_title",
"install"
)

1
openpype/hosts/testhost/api/context.json
View file

@ -1 +0,0 @@
{}

108
openpype/hosts/testhost/api/instances.json
View file

@ -1,108 +0,0 @@
[
{
"id": "pyblish.avalon.instance",
"active": true,
"family": "test",
"subset": "testMyVariant",
"version": 1,
"asset": "sq01_sh0010",
"task": "Compositing",
"variant": "myVariant",
"instance_id": "a485f148-9121-46a5-8157-aa64df0fb449",
"creator_attributes": {
"number_key": 10,
"ha": 10
},
"publish_attributes": {
"CollectFtrackApi": {
"add_ftrack_family": false
}
},
"creator_identifier": "test_one"
},
{
"id": "pyblish.avalon.instance",
"active": true,
"family": "test",
"subset": "testMyVariant2",
"version": 1,
"asset": "sq01_sh0010",
"task": "Compositing",
"variant": "myVariant2",
"creator_attributes": {},
"instance_id": "a485f148-9121-46a5-8157-aa64df0fb444",
"publish_attributes": {
"CollectFtrackApi": {
"add_ftrack_family": true
}
},
"creator_identifier": "test_one"
},
{
"id": "pyblish.avalon.instance",
"active": true,
"family": "test",
"subset": "testMain",
"version": 1,
"asset": "sq01_sh0010",
"task": "Compositing",
"variant": "Main",
"creator_attributes": {},
"instance_id": "3607bc95-75f6-4648-a58d-e699f413d09f",
"publish_attributes": {
"CollectFtrackApi": {
"add_ftrack_family": true
}
},
"creator_identifier": "test_two"
},
{
"id": "pyblish.avalon.instance",
"active": true,
"family": "test",
"subset": "testMain2",
"version": 1,
"asset": "sq01_sh0020",
"task": "Compositing",
"variant": "Main2",
"instance_id": "4ccf56f6-9982-4837-967c-a49695dbe8eb",
"creator_attributes": {},
"publish_attributes": {
"CollectFtrackApi": {
"add_ftrack_family": true
}
},
"creator_identifier": "test_two"
},
{
"id": "pyblish.avalon.instance",
"family": "test_three",
"subset": "test_threeMain2",
"active": true,
"version": 1,
"asset": "sq01_sh0020",
"task": "Compositing",
"variant": "Main2",
"instance_id": "4ccf56f6-9982-4837-967c-a49695dbe8ec",
"creator_attributes": {},
"publish_attributes": {
"CollectFtrackApi": {
"add_ftrack_family": true
}
}
},
{
"id": "pyblish.avalon.instance",
"family": "workfile",
"subset": "workfileMain",
"active": true,
"creator_identifier": "workfile",
"version": 1,
"asset": "Alpaca_01",
"task": "modeling",
"variant": "Main",
"instance_id": "7c9ddfc7-9f9c-4c1c-b233-38c966735fb6",
"creator_attributes": {},
"publish_attributes": {}
}
]

155
openpype/hosts/testhost/api/pipeline.py
View file

@ -1,155 +0,0 @@
import os
import json
from openpype.client import get_asset_by_name
class HostContext:
instances_json_path = None
context_json_path = None
@classmethod
def get_context_title(cls):
project_name = os.environ.get("AVALON_PROJECT")
if not project_name:
return "TestHost"
asset_name = os.environ.get("AVALON_ASSET")
if not asset_name:
return project_name
asset_doc = get_asset_by_name(
project_name, asset_name, fields=["data.parents"]
)
parents = asset_doc.get("data", {}).get("parents") or []
hierarchy = [project_name]
hierarchy.extend(parents)
hierarchy.append("<b>{}</b>".format(asset_name))
task_name = os.environ.get("AVALON_TASK")
if task_name:
hierarchy.append(task_name)
return "/".join(hierarchy)
@classmethod
def get_current_dir_filepath(cls, filename):
return os.path.join(
os.path.dirname(os.path.abspath(__file__)),
filename
)
@classmethod
def get_instances_json_path(cls):
if cls.instances_json_path is None:
cls.instances_json_path = cls.get_current_dir_filepath(
"instances.json"
)
return cls.instances_json_path
@classmethod
def get_context_json_path(cls):
if cls.context_json_path is None:
cls.context_json_path = cls.get_current_dir_filepath(
"context.json"
)
return cls.context_json_path
@classmethod
def add_instance(cls, instance):
instances = cls.get_instances()
instances.append(instance)
cls.save_instances(instances)
@classmethod
def save_instances(cls, instances):
json_path = cls.get_instances_json_path()
with open(json_path, "w") as json_stream:
json.dump(instances, json_stream, indent=4)
@classmethod
def get_instances(cls):
json_path = cls.get_instances_json_path()
if not os.path.exists(json_path):
instances = []
with open(json_path, "w") as json_stream:
json.dump(json_stream, instances)
else:
with open(json_path, "r") as json_stream:
instances = json.load(json_stream)
return instances
@classmethod
def get_context_data(cls):
json_path = cls.get_context_json_path()
if not os.path.exists(json_path):
data = {}
with open(json_path, "w") as json_stream:
json.dump(data, json_stream)
else:
with open(json_path, "r") as json_stream:
data = json.load(json_stream)
return data
@classmethod
def save_context_data(cls, data):
json_path = cls.get_context_json_path()
with open(json_path, "w") as json_stream:
json.dump(data, json_stream, indent=4)
def ls():
return []
def list_instances():
return HostContext.get_instances()
def update_instances(update_list):
updated_instances = {}
for instance, _changes in update_list:
updated_instances[instance.id] = instance.data_to_store()
instances = HostContext.get_instances()
for instance_data in instances:
instance_id = instance_data["instance_id"]
if instance_id in updated_instances:
new_instance_data = updated_instances[instance_id]
old_keys = set(instance_data.keys())
new_keys = set(new_instance_data.keys())
instance_data.update(new_instance_data)
for key in (old_keys - new_keys):
instance_data.pop(key)
HostContext.save_instances(instances)
def remove_instances(instances):
if not isinstance(instances, (tuple, list)):
instances = [instances]
current_instances = HostContext.get_instances()
for instance in instances:
instance_id = instance.data["instance_id"]
found_idx = None
for idx, _instance in enumerate(current_instances):
if instance_id == _instance["instance_id"]:
found_idx = idx
break
if found_idx is not None:
current_instances.pop(found_idx)
HostContext.save_instances(current_instances)
def get_context_data():
return HostContext.get_context_data()
def update_context_data(data, changes):
HostContext.save_context_data(data)
def get_context_title():
return HostContext.get_context_title()

75
openpype/hosts/testhost/plugins/create/auto_creator.py
View file

@ -1,75 +0,0 @@
from openpype.lib import NumberDef
from openpype.client import get_asset_by_name
from openpype.pipeline import (
legacy_io,
AutoCreator,
CreatedInstance,
)
from openpype.hosts.testhost.api import pipeline
class MyAutoCreator(AutoCreator):
identifier = "workfile"
family = "workfile"
def get_instance_attr_defs(self):
output = [
NumberDef("number_key", label="Number")
]
return output
def collect_instances(self):
for instance_data in pipeline.list_instances():
creator_id = instance_data.get("creator_identifier")
if creator_id == self.identifier:
subset_name = instance_data["subset"]
instance = CreatedInstance(
self.family, subset_name, instance_data, self
)
self._add_instance_to_context(instance)
def update_instances(self, update_list):
pipeline.update_instances(update_list)
def create(self):
existing_instance = None
for instance in self.create_context.instances:
if instance.family == self.family:
existing_instance = instance
break
variant = "Main"
project_name = legacy_io.Session["AVALON_PROJECT"]
asset_name = legacy_io.Session["AVALON_ASSET"]
task_name = legacy_io.Session["AVALON_TASK"]
host_name = legacy_io.Session["AVALON_APP"]
if existing_instance is None:
asset_doc = get_asset_by_name(project_name, asset_name)
subset_name = self.get_subset_name(
variant, task_name, asset_doc, project_name, host_name
)
data = {
"asset": asset_name,
"task": task_name,
"variant": variant
}
data.update(self.get_dynamic_data(
variant, task_name, asset_doc, project_name, host_name
))
new_instance = CreatedInstance(
self.family, subset_name, data, self
)
self._add_instance_to_context(new_instance)
elif (
existing_instance["asset"] != asset_name
or existing_instance["task"] != task_name
):
asset_doc = get_asset_by_name(project_name, asset_name)
subset_name = self.get_subset_name(
variant, task_name, asset_doc, project_name, host_name
)
existing_instance["asset"] = asset_name
existing_instance["task"] = task_name

94
openpype/hosts/testhost/plugins/create/test_creator_1.py
View file

@ -1,94 +0,0 @@
import json
from openpype import resources
from openpype.hosts.testhost.api import pipeline
from openpype.lib import (
UISeparatorDef,
UILabelDef,
BoolDef,
NumberDef,
FileDef,
)
from openpype.pipeline import (
Creator,
CreatedInstance,
)
class TestCreatorOne(Creator):
identifier = "test_one"
label = "test"
family = "test"
description = "Testing creator of testhost"
create_allow_context_change = False
def get_icon(self):
return resources.get_openpype_splash_filepath()
def collect_instances(self):
for instance_data in pipeline.list_instances():
creator_id = instance_data.get("creator_identifier")
if creator_id == self.identifier:
instance = CreatedInstance.from_existing(
instance_data, self
)
self._add_instance_to_context(instance)
def update_instances(self, update_list):
pipeline.update_instances(update_list)
def remove_instances(self, instances):
pipeline.remove_instances(instances)
for instance in instances:
self._remove_instance_from_context(instance)
def create(self, subset_name, data, pre_create_data):
print("Data that can be used in create:\n{}".format(
json.dumps(pre_create_data, indent=4)
))
new_instance = CreatedInstance(self.family, subset_name, data, self)
pipeline.HostContext.add_instance(new_instance.data_to_store())
self.log.info(new_instance.data)
self._add_instance_to_context(new_instance)
def get_default_variants(self):
return [
"myVariant",
"variantTwo",
"different_variant"
]
def get_instance_attr_defs(self):
output = [
NumberDef("number_key", label="Number"),
]
return output
def get_pre_create_attr_defs(self):
output = [
BoolDef("use_selection", label="Use selection"),
UISeparatorDef(),
UILabelDef("Testing label"),
FileDef("filepath", folders=True, label="Filepath"),
FileDef(
"filepath_2", multipath=True, folders=True, label="Filepath 2"
)
]
return output
def get_detail_description(self):
return """# Relictus funes est Nyseides currusque nunc oblita
## Causa sed
Lorem markdownum posito consumptis, *plebe Amorque*, abstitimus rogatus fictaque
gladium Circe, nos? Bos aeternum quae. Utque me, si aliquem cladis, et vestigia
arbor, sic mea ferre lacrimae agantur prospiciens hactenus. Amanti dentes pete,
vos quid laudemque rastrorumque terras in gratantibus **radix** erat cedemus?
Pudor tu ponderibus verbaque illa; ire ergo iam Venus patris certe longae
cruentum lecta, et quaeque. Sit doce nox. Anteit ad tempora magni plenaque et
videres mersit sibique auctor in tendunt mittit cunctos ventisque gravitate
volucris quemquam Aeneaden. Pectore Mensis somnus; pectora
[ferunt](http://www.mox.org/oculosbracchia)? Fertilitatis bella dulce et suum?
"""

74
openpype/hosts/testhost/plugins/create/test_creator_2.py
View file

@ -1,74 +0,0 @@
from openpype.lib import NumberDef, TextDef
from openpype.hosts.testhost.api import pipeline
from openpype.pipeline import (
Creator,
CreatedInstance,
)
class TestCreatorTwo(Creator):
identifier = "test_two"
label = "test"
family = "test"
description = "A second testing creator"
def get_icon(self):
return "cube"
def create(self, subset_name, data, pre_create_data):
new_instance = CreatedInstance(self.family, subset_name, data, self)
pipeline.HostContext.add_instance(new_instance.data_to_store())
self.log.info(new_instance.data)
self._add_instance_to_context(new_instance)
def collect_instances(self):
for instance_data in pipeline.list_instances():
creator_id = instance_data.get("creator_identifier")
if creator_id == self.identifier:
instance = CreatedInstance.from_existing(
instance_data, self
)
self._add_instance_to_context(instance)
def update_instances(self, update_list):
pipeline.update_instances(update_list)
def remove_instances(self, instances):
pipeline.remove_instances(instances)
for instance in instances:
self._remove_instance_from_context(instance)
def get_instance_attr_defs(self):
output = [
NumberDef("number_key"),
TextDef("text_key")
]
return output
def get_detail_description(self):
return """# Lorem ipsum, dolor sit amet. [![Awesome](https://cdn.rawgit.com/sindresorhus/awesome/d7305f38d29fed78fa85652e3a63e154dd8e8829/media/badge.svg)](https://github.com/sindresorhus/awesome)
> A curated list of awesome lorem ipsum generators.
Inspired by the [awesome](https://github.com/sindresorhus/awesome) list thing.
## Table of Contents
- [Legend](#legend)
- [Practical](#briefcase-practical)
- [Whimsical](#roller_coaster-whimsical)
- [Animals](#rabbit-animals)
- [Eras](#tophat-eras)
- [Famous Individuals](#sunglasses-famous-individuals)
- [Music](#microphone-music)
- [Food and Drink](#pizza-food-and-drink)
- [Geographic and Dialects](#earth_africa-geographic-and-dialects)
- [Literature](#books-literature)
- [Miscellaneous](#cyclone-miscellaneous)
- [Sports and Fitness](#bicyclist-sports-and-fitness)
- [TV and Film](#movie_camera-tv-and-film)
- [Tools, Apps, and Extensions](#wrench-tools-apps-and-extensions)
- [Contribute](#contribute)
- [TODO](#todo)
"""

34
openpype/hosts/testhost/plugins/publish/collect_context.py
View file

@ -1,34 +0,0 @@
import pyblish.api
from openpype.pipeline import (
OpenPypePyblishPluginMixin,
attribute_definitions
)
class CollectContextDataTestHost(
pyblish.api.ContextPlugin, OpenPypePyblishPluginMixin
):
"""
Collecting temp json data sent from a host context
and path for returning json data back to hostself.
"""
label = "Collect Source - Test Host"
order = pyblish.api.CollectorOrder - 0.4
hosts = ["testhost"]
@classmethod
def get_attribute_defs(cls):
return [
attribute_definitions.BoolDef(
"test_bool",
True,
label="Bool input"
)
]
def process(self, context):
# get json paths from os and load them
for instance in context:
instance.data["source"] = "testhost"

52
openpype/hosts/testhost/plugins/publish/collect_instance_1.py
View file

@ -1,52 +0,0 @@
import json
import pyblish.api
from openpype.lib import attribute_definitions
from openpype.pipeline import OpenPypePyblishPluginMixin
class CollectInstanceOneTestHost(
pyblish.api.InstancePlugin, OpenPypePyblishPluginMixin
):
"""
Collecting temp json data sent from a host context
and path for returning json data back to hostself.
"""
label = "Collect Instance 1 - Test Host"
order = pyblish.api.CollectorOrder - 0.3
hosts = ["testhost"]
@classmethod
def get_attribute_defs(cls):
return [
attribute_definitions.NumberDef(
"version",
default=1,
minimum=1,
maximum=999,
decimals=0,
label="Version"
)
]
def process(self, instance):
self._debug_log(instance)
publish_attributes = instance.data.get("publish_attributes")
if not publish_attributes:
return
values = publish_attributes.get(self.__class__.__name__)
if not values:
return
instance.data["version"] = values["version"]
def _debug_log(self, instance):
def _default_json(value):
return str(value)
self.log.info(
json.dumps(instance.data, indent=4, default=_default_json)
)

57
openpype/hosts/testhost/plugins/publish/validate_context_with_error.py
View file

@ -1,57 +0,0 @@
import pyblish.api
from openpype.pipeline import PublishValidationError
class ValidateInstanceAssetRepair(pyblish.api.Action):
"""Repair the instance asset."""
label = "Repair"
icon = "wrench"
on = "failed"
def process(self, context, plugin):
pass
description = """
## Publish plugins
### Validate Scene Settings
#### Skip Resolution Check for Tasks
Set regex pattern(s) to look for in a Task name to skip resolution check against values from DB.
#### Skip Timeline Check for Tasks
Set regex pattern(s) to look for in a Task name to skip `frameStart`, `frameEnd` check against values from DB.
### AfterEffects Submit to Deadline
* `Use Published scene` - Set to True (green) when Deadline should take published scene as a source instead of uploaded local one.
* `Priority` - priority of job on farm
* `Primary Pool` - here is list of pool fetched from server you can select from.
* `Secondary Pool`
* `Frames Per Task` - number of sequence division between individual tasks (chunks)
making one job on farm.
"""
class ValidateContextWithError(pyblish.api.ContextPlugin):
"""Validate the instance asset is the current selected context asset.
As it might happen that multiple worfiles are opened, switching
between them would mess with selected context.
In that case outputs might be output under wrong asset!
Repair action will use Context asset value (from Workfiles or Launcher)
Closing and reopening with Workfiles will refresh Context value.
"""
label = "Validate Context With Error"
hosts = ["testhost"]
actions = [ValidateInstanceAssetRepair]
order = pyblish.api.ValidatorOrder
def process(self, context):
raise PublishValidationError("Crashing", "Context error", description)

57
openpype/hosts/testhost/plugins/publish/validate_with_error.py
View file

@ -1,57 +0,0 @@
import pyblish.api
from openpype.pipeline import PublishValidationError
class ValidateInstanceAssetRepair(pyblish.api.Action):
"""Repair the instance asset."""
label = "Repair"
icon = "wrench"
on = "failed"
def process(self, context, plugin):
pass
description = """
## Publish plugins
### Validate Scene Settings
#### Skip Resolution Check for Tasks
Set regex pattern(s) to look for in a Task name to skip resolution check against values from DB.
#### Skip Timeline Check for Tasks
Set regex pattern(s) to look for in a Task name to skip `frameStart`, `frameEnd` check against values from DB.
### AfterEffects Submit to Deadline
* `Use Published scene` - Set to True (green) when Deadline should take published scene as a source instead of uploaded local one.
* `Priority` - priority of job on farm
* `Primary Pool` - here is list of pool fetched from server you can select from.
* `Secondary Pool`
* `Frames Per Task` - number of sequence division between individual tasks (chunks)
making one job on farm.
"""
class ValidateWithError(pyblish.api.InstancePlugin):
"""Validate the instance asset is the current selected context asset.
As it might happen that multiple worfiles are opened, switching
between them would mess with selected context.
In that case outputs might be output under wrong asset!
Repair action will use Context asset value (from Workfiles or Launcher)
Closing and reopening with Workfiles will refresh Context value.
"""
label = "Validate With Error"
hosts = ["testhost"]
actions = [ValidateInstanceAssetRepair]
order = pyblish.api.ValidatorOrder
def process(self, instance):
raise PublishValidationError("Crashing", "Instance error", description)

68
openpype/hosts/testhost/run_publish.py
View file

@ -1,68 +0,0 @@
import os
import sys
mongo_url = ""
project_name = ""
asset_name = ""
task_name = ""
ftrack_url = ""
ftrack_username = ""
ftrack_api_key = ""
def multi_dirname(path, times=1):
for _ in range(times):
path = os.path.dirname(path)
return path
host_name = "testhost"
current_file = os.path.abspath(__file__)
openpype_dir = multi_dirname(current_file, 4)
os.environ["OPENPYPE_MONGO"] = mongo_url
os.environ["OPENPYPE_ROOT"] = openpype_dir
os.environ["AVALON_PROJECT"] = project_name
os.environ["AVALON_ASSET"] = asset_name
os.environ["AVALON_TASK"] = task_name
os.environ["AVALON_APP"] = host_name
os.environ["OPENPYPE_DATABASE_NAME"] = "openpype"
os.environ["AVALON_TIMEOUT"] = "1000"
os.environ["AVALON_DB"] = "avalon"
os.environ["FTRACK_SERVER"] = ftrack_url
os.environ["FTRACK_API_USER"] = ftrack_username
os.environ["FTRACK_API_KEY"] = ftrack_api_key
for path in [
openpype_dir,
r"{}\repos\avalon-core".format(openpype_dir),
r"{}\.venv\Lib\site-packages".format(openpype_dir)
]:
sys.path.append(path)
from Qt import QtWidgets, QtCore
from openpype.tools.publisher.window import PublisherWindow
def main():
"""Main function for testing purposes."""
import pyblish.api
from openpype.pipeline import install_host
from openpype.modules import ModulesManager
from openpype.hosts.testhost import api as testhost
manager = ModulesManager()
for plugin_path in manager.collect_plugin_paths()["publish"]:
pyblish.api.register_plugin_path(plugin_path)
install_host(testhost)
QtWidgets.QApplication.setAttribute(QtCore.Qt.AA_EnableHighDpiScaling)
app = QtWidgets.QApplication([])
window = PublisherWindow()
window.show()
app.exec_()
if __name__ == "__main__":
main()