Farm publishing: check if published items do exist

openpype/cli.py

@@ -115,15 +115,17 @@ def extractenvironments(output_json_path, project, asset, task, app):
 @main.command()
 @click.argument("paths", nargs=-1)
 @click.option("-d", "--debug", is_flag=True, help="Print debug messages")
-def publish(debug, paths):
+@click.option("-t", "--targets", help="Targets module", default=None,
+ multiple=True)
+def publish(debug, paths, targets):
  """Start CLI publishing.
 
  Publish collects json from paths provided as an argument.
  More than one path is allowed.
  """
  if debug:
  os.environ['OPENPYPE_DEBUG'] = '3'
- PypeCommands.publish(list(paths))
+ PypeCommands.publish(list(paths), targets)
 
 
 @main.command()

openpype/lib/abstract_submit_deadline.py

@@ -18,6 +18,48 @@
 from .abstract_metaplugins import AbstractMetaInstancePlugin
 
 
+def requests_post(*args, **kwargs):
+ """Wrap request post method.
+
+ Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
+ variable is found. This is useful when Deadline or Muster server are
+ running with self-signed certificates and their certificate is not
+ added to trusted certificates on client machines.
+
+ Warning:
+ Disabling SSL certificate validation is defeating one line
+ of defense SSL is providing and it is not recommended.
+
+ """
+ if 'verify' not in kwargs:
+ kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL",
+ True) else True # noqa
+ # add 10sec timeout before bailing out
+ kwargs['timeout'] = 10
+ return requests.post(*args, **kwargs)
+
+
+def requests_get(*args, **kwargs):
+ """Wrap request get method.
+
+ Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
+ variable is found. This is useful when Deadline or Muster server are
+ running with self-signed certificates and their certificate is not
+ added to trusted certificates on client machines.
+
+ Warning:
+ Disabling SSL certificate validation is defeating one line
+ of defense SSL is providing and it is not recommended.
+
+ """
+ if 'verify' not in kwargs:
+ kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL",
+ True) else True # noqa
+ # add 10sec timeout before bailing out
+ kwargs['timeout'] = 10
+ return requests.get(*args, **kwargs)
+
+
 @attr.s
 class DeadlineJobInfo(object):
  """Mapping of all Deadline *JobInfo* attributes.
@@ -579,7 +621,7 @@ def submit(self, payload):
 
  """
  url = "{}/api/jobs".format(self._deadline_url)
- response = self._requests_post(url, json=payload)
+ response = requests_post(url, json=payload)
  if not response.ok:
  self.log.error("Submission failed!")
  self.log.error(response.status_code)
@@ -592,41 +634,3 @@ def submit(self, payload):
  self._instance.data["deadlineSubmissionJob"] = result
 
  return result["_id"]
-
- def _requests_post(self, *args, **kwargs):
- """Wrap request post method.
-
- Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
- variable is found. This is useful when Deadline or Muster server are
- running with self-signed certificates and their certificate is not
- added to trusted certificates on client machines.
-
- Warning:
- Disabling SSL certificate validation is defeating one line
- of defense SSL is providing and it is not recommended.
-
- """
- if 'verify' not in kwargs:
- kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL", True) else True # noqa
- # add 10sec timeout before bailing out
- kwargs['timeout'] = 10
- return requests.post(*args, **kwargs)
-
- def _requests_get(self, *args, **kwargs):
- """Wrap request get method.
-
- Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
- variable is found. This is useful when Deadline or Muster server are
- running with self-signed certificates and their certificate is not
- added to trusted certificates on client machines.
-
- Warning:
- Disabling SSL certificate validation is defeating one line
- of defense SSL is providing and it is not recommended.
-
- """
- if 'verify' not in kwargs:
- kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL", True) else True # noqa
- # add 10sec timeout before bailing out
- kwargs['timeout'] = 10
- return requests.get(*args, **kwargs)

openpype/modules/deadline/plugins/publish/submit_publish_job.py

@@ -231,7 +231,8 @@ def _submit_deadline_post_job(self, instance, job, instances):
 
  args = [
  'publish',
- roothless_metadata_path
+ roothless_metadata_path,
+ "--targets {}".format("deadline")
  ]
 
  # Generate the payload for Deadline submission

openpype/modules/deadline/plugins/publish/validate_expected_and_rendered_files.py

@@ -0,0 +1,186 @@
+import os
+import json
+import pyblish.api
+
+from avalon.vendor import requests
+
+from openpype.api import get_system_settings
+from openpype.lib.abstract_submit_deadline import requests_get
+from openpype.lib.delivery import collect_frames
+
+
+class ValidateExpectedFiles(pyblish.api.InstancePlugin):
+ """Compare rendered and expected files"""
+
+ label = "Validate rendered files from Deadline"
+ order = pyblish.api.ValidatorOrder
+ families = ["render"]
+ targets = ["deadline"]
+
+ # check if actual frame range on render job wasn't different
+ # case when artists wants to render only subset of frames
+ allow_user_override = True
+
+ def process(self, instance):
+ frame_list = self._get_frame_list(instance.data["render_job_id"])
+
+ for repre in instance.data["representations"]:
+ expected_files = self._get_expected_files(repre)
+
+ staging_dir = repre["stagingDir"]
+ existing_files = self._get_existing_files(staging_dir)
+
+ expected_non_existent = expected_files.difference(
+ existing_files)
+ if len(expected_non_existent) != 0:
+ self.log.info("Some expected files missing {}".format(
+ expected_non_existent))
+
+ if self.allow_user_override:
+ file_name_template, frame_placeholder = \
+ self._get_file_name_template_and_placeholder(
+ expected_files)
+
+ if not file_name_template:
+ return
+
+ real_expected_rendered = self._get_real_render_expected(
+ file_name_template,
+ frame_placeholder,
+ frame_list)
+
+ real_expected_non_existent = \
+ real_expected_rendered.difference(existing_files)
+ if len(real_expected_non_existent) != 0:
+ raise RuntimeError("Still missing some files {}".
+ format(real_expected_non_existent))
+ self.log.info("Update range from actual job range")
+ repre["files"] = sorted(list(real_expected_rendered))
+ else:
+ raise RuntimeError("Some expected files missing {}".format(
+ expected_non_existent))
+
+ def _get_frame_list(self, original_job_id):
+ """
+ Returns list of frame ranges from all render job.
+
+ Render job might be requeried so job_id in metadata.json is invalid
+ GlobalJobPreload injects current ids to RENDER_JOB_IDS.
+
+ Args:
+ original_job_id (str)
+ Returns:
+ (list)
+ """
+ all_frame_lists = []
+ render_job_ids = os.environ.get("RENDER_JOB_IDS")
+ if render_job_ids:
+ render_job_ids = render_job_ids.split(',')
+ else: # fallback
+ render_job_ids = [original_job_id]
+
+ for job_id in render_job_ids:
+ job_info = self._get_job_info(job_id)
+ frame_list = job_info["Props"]["Frames"]
+ if frame_list:
+ all_frame_lists.extend(frame_list.split(','))
+
+ return all_frame_lists
+
+ def _get_real_render_expected(self, file_name_template, frame_placeholder,
+ frame_list):
+ """
+ Calculates list of names of expected rendered files.
+
+ Might be different from job expected files if user explicitly and
+ manually change frame list on Deadline job.
+ """
+ real_expected_rendered = set()
+ src_padding_exp = "%0{}d".format(len(frame_placeholder))
+ for frames in frame_list:
+ if '-' not in frames: # single frame
+ frames = "{}-{}".format(frames, frames)
+
+ start, end = frames.split('-')
+ for frame in range(int(start), int(end) + 1):
+ ren_name = file_name_template.replace(
+ frame_placeholder, src_padding_exp % frame)
+ real_expected_rendered.add(ren_name)
+
+ return real_expected_rendered
+
+ def _get_file_name_template_and_placeholder(self, files):
+ """Returns file name with frame replaced with # and this placeholder"""
+ sources_and_frames = collect_frames(files)
+
+ file_name_template = frame_placeholder = None
+ for file_name, frame in sources_and_frames.items():
+ frame_placeholder = "#" * len(frame)
+ file_name_template = os.path.basename(
+ file_name.replace(frame, frame_placeholder))
+ break
+
+ return file_name_template, frame_placeholder
+
+ def _get_job_info(self, job_id):
+ """
+ Calls DL for actual job info for 'job_id'
+
+ Might be different than job info saved in metadata.json if user
+ manually changes job pre/during rendering.
+ """
+ deadline_url = (
+ get_system_settings()
+ ["modules"]
+ ["deadline"]
+ ["DEADLINE_REST_URL"]
+ )
+ assert deadline_url, "Requires DEADLINE_REST_URL"
+
+ url = "{}/api/jobs?JobID={}".format(deadline_url, job_id)
+ try:
+ response = requests_get(url)
+ except requests.exceptions.ConnectionError:
+ print("Deadline is not accessible at {}".format(deadline_url))
+ # self.log("Deadline is not accessible at {}".format(deadline_url))
+ return {}
+
+ if not response.ok:
+ self.log.error("Submission failed!")
+ self.log.error(response.status_code)
+ self.log.error(response.content)
+ raise RuntimeError(response.text)
+
+ json_content = response.json()
+ if json_content:
+ return json_content.pop()
+ return {}
+
+ def _parse_metadata_json(self, json_path):
+ if not os.path.exists(json_path):
+ msg = "Metadata file {} doesn't exist".format(json_path)
+ raise RuntimeError(msg)
+
+ with open(json_path) as fp:
+ try:
+ return json.load(fp)
+ except Exception as exc:
+ self.log.error(
+ "Error loading json: "
+ "{} - Exception: {}".format(json_path, exc)
+ )
+
+ def _get_existing_files(self, out_dir):
+ """Returns set of existing file names from 'out_dir'"""
+ existing_files = set()
+ for file_name in os.listdir(out_dir):
+ existing_files.add(file_name)
+ return existing_files
+
+ def _get_expected_files(self, repre):
+ """Returns set of file names from metadata.json"""
+ expected_files = set()
+
+ for file_name in repre["files"]:
+ expected_files.add(file_name)
+ return expected_files

openpype/plugins/publish/collect_rendered_files.py

@@ -87,11 +87,14 @@ def _process_path(self, data, anatomy):
  instance = self._context.create_instance(
  instance_data.get("subset")
  )
- self.log.info("Filling stagignDir...")
+ self.log.info("Filling stagingDir...")
 
  self._fill_staging_dir(instance_data, anatomy)
  instance.data.update(instance_data)
 
+ # stash render job id for later validation
+ instance.data["render_job_id"] = data.get("job").get("_id")
+
  representations = []
  for repre_data in instance_data.get("representations") or []:
  self._fill_staging_dir(repre_data, anatomy)

openpype/pype_commands.py

@@ -46,16 +46,18 @@ def launch_standalone_publisher():
  standalonepublish.main()
 
  @staticmethod
- def publish(paths):
+ def publish(paths, targets=None):
  """Start headless publishing.
 
  Publish use json from passed paths argument.
 
  Args:
  paths (list): Paths to jsons.
+ targets (string): What module should be targeted
+ (to choose validator for example)
 
  Raises:
- RuntimeError: When there is no pathto process.
+ RuntimeError: When there is no path to process.
  """
  if not any(paths):
  raise RuntimeError("No publish paths specified")
@@ -82,6 +84,10 @@ def publish(paths):
  pyblish.api.register_target("filesequence")
  pyblish.api.register_host("shell")
 
+ if targets:
+ for target in targets:
+ pyblish.api.register_target(target)
+
  os.environ["OPENPYPE_PUBLISH_DATA"] = os.pathsep.join(paths)
 
  log.info("Running publish ...")

openpype/settings/defaults/project_settings/deadline.json

@@ -1,5 +1,12 @@
 {
  "publish": {
+ "ValidateExpectedFiles": {
+ "enabled": true,
+ "active": true,
+ "families": ["render"],
+ "targets": ["deadline"],
+ "allow_user_override": true
+ },
  "MayaSubmitDeadline": {
  "enabled": true,
  "optional": false,