Added support for custom adapter hooks

This adds support for attributing custom hooks to adapters and executing them with `hook_function_argument_map` being passed along through the adapter IO functions.

Signed-off-by: Tim Lehr <[email protected]>

src/py-opentimelineio/opentimelineio/adapters/adapter.py

@@ -10,6 +10,7 @@
 import inspect
 import collections
 import copy
+from typing import List
 
 from .. import (
  core,
@@ -99,6 +100,21 @@ def read_from_file(
  media_linker_argument_map or {}
  )
 
+ hook_function_argument_map = copy.deepcopy(
+ hook_function_argument_map or {}
+ )
+ hook_function_argument_map['adapter_arguments'] = copy.deepcopy(
+ adapter_argument_map
+ )
+ hook_function_argument_map['media_linker_argument_map'] = (
+ media_linker_argument_map
+ )
+
+ if self.has_feature("hooks"):
+ adapter_argument_map[
+ "hook_function_argument_map"
+ ] = hook_function_argument_map
+
  result = None
 
  if (
@@ -119,15 +135,6 @@ def read_from_file(
  **adapter_argument_map
  )
 
- hook_function_argument_map = copy.deepcopy(
- hook_function_argument_map or {}
- )
- hook_function_argument_map['adapter_arguments'] = copy.deepcopy(
- adapter_argument_map
- )
- hook_function_argument_map['media_linker_argument_map'] = (
- media_linker_argument_map
- )
  result = hooks.run(
  "post_adapter_read",
  result,
@@ -174,6 +181,11 @@ def write_to_file(
  # Store file path for use in hooks
  hook_function_argument_map['_filepath'] = filepath
 
+ if self.has_feature("hooks"):
+ adapter_argument_map[
+ "hook_function_argument_map"
+ ] = hook_function_argument_map
+
  input_otio = hooks.run("pre_adapter_write", input_otio,
  extra_args=hook_function_argument_map)
  if (
@@ -210,13 +222,6 @@ def read_from_string(
  **adapter_argument_map
  ):
  """Call the read_from_string function on this adapter."""
-
- result = self._execute_function(
- "read_from_string",
- input_str=input_str,
- **adapter_argument_map
- )
-
  hook_function_argument_map = copy.deepcopy(
  hook_function_argument_map or {}
  )
@@ -227,6 +232,17 @@ def read_from_string(
  media_linker_argument_map
  )
 
+ if self.has_feature("hooks"):
+ adapter_argument_map[
+ "hook_function_argument_map"
+ ] = hook_function_argument_map
+
+ result = self._execute_function(
+ "read_from_string",
+ input_str=input_str,
+ **adapter_argument_map
+ )
+
  result = hooks.run(
  "post_adapter_read",
  result,
@@ -277,6 +293,16 @@ def write_to_string(
  **adapter_argument_map
  )
 
+ def adapter_hook_names(self) -> List[str]:
+ """Returns a list of hooks claimed by the adapter.
+
+ In addition to the hook being declared in the manifest, it should also be
+ returned here, so it can be attributed to the adapter.
+ """
+ if not self.has_feature("hooks"):
+ return []
+ return self._execute_function("adapter_hook_names")
+
  def __str__(self):
  return (
  "Adapter("
@@ -372,5 +398,6 @@ def _with_linked_media_references(
  'read': ['read_from_file', 'read_from_string'],
  'write_to_file': ['write_to_file'],
  'write_to_string': ['write_to_string'],
- 'write': ['write_to_file', 'write_to_string']
+ 'write': ['write_to_file', 'write_to_string'],
+ 'hooks': ['adapter_hook_names']
 }

tests/baselines/adapter_plugin_manifest.plugin_manifest.json

@@ -16,13 +16,17 @@
  },
  {
  "FROM_TEST_FILE" : "post_write_hookscript_example.json"
+ },
+ {
+ "FROM_TEST_FILE" : "custom_adapter_hookscript_example.json"
  }
  ],
  "hooks" : {
  "pre_adapter_write" : ["example hook", "example hook"],
  "post_adapter_read" : [],
  "post_adapter_write" : ["post write example hook"],
- "post_media_linker" : ["example hook"]
+ "post_media_linker" : ["example hook"],
+ "custom_adapter_hook": ["custom adapter hook"]
  },
  "version_manifests" : {
  "TEST_FAMILY_NAME": {

tests/baselines/custom_adapter_hookscript_example.json

@@ -0,0 +1,5 @@
+{
+ "OTIO_SCHEMA" : "HookScript.1",
+ "name" : "custom adapter hook",
+ "filepath" : "custom_adapter_hookscript_example.py"
+}

tests/baselines/custom_adapter_hookscript_example.py

@@ -0,0 +1,13 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright Contributors to the OpenTimelineIO project
+
+"""This file is here to support the test_adapter_plugin unittest, specifically adapters
+that implement their own hooks.
+If you want to learn how to write your own adapter plugin, please read:
+https://opentimelineio.readthedocs.io/en/latest/tutorials/write-an-adapter.html
+"""
+
+
+def hook_function(in_timeline, argument_map=None):
+ in_timeline.metadata["custom_hook"] = dict(argument_map)
+ return in_timeline

tests/baselines/example.py

@@ -6,23 +6,41 @@
 https://opentimelineio.readthedocs.io/en/latest/tutorials/write-an-adapter.html
 """
 
-import opentimelineio as otio
 
+# `hook_function_argument_map` is only a required argument for adapters that implement
+# custom hooks.
+def read_from_file(filepath, suffix="", hook_function_argument_map=None):
+ import opentimelineio as otio
 
-def read_from_file(filepath, suffix=""):
  fake_tl = otio.schema.Timeline(name=filepath + str(suffix))
  fake_tl.tracks.append(otio.schema.Track())
  fake_tl.tracks[0].append(otio.schema.Clip(name=filepath + "_clip"))
+
+ if (hook_function_argument_map and
+ hook_function_argument_map.get("run_custom_hook", False)):
+ return otio.hooks.run(hook="custom_adapter_hook", tl=fake_tl,
+ extra_args=hook_function_argument_map)
+
  return fake_tl
 
 
-def read_from_string(input_str, suffix=""):
- return read_from_file(input_str, suffix)
+# `hook_function_argument_map` is only a required argument for adapters that implement
+# custom hooks.
+def read_from_string(input_str, suffix="", hook_function_argument_map=None):
+ tl = read_from_file(input_str, suffix, hook_function_argument_map)
+ return tl
+
+
+# this is only required for adapters that implement custom hooks
+def adapter_hook_names():
+ return ["custom_adapter_hook"]
 
 
 # in practice, these will be in separate plugins, but for simplicity in the
 # unit tests, we put this in the same file as the example adapter.
 def link_media_reference(in_clip, media_linker_argument_map):
+ import opentimelineio as otio
+
  d = {'from_test_linker': True}
  d.update(media_linker_argument_map)
  return otio.schema.MissingReference(

tests/test_adapter_plugin.py

@@ -89,6 +89,7 @@ def test_has_feature(self):
  self.assertTrue(self.adp.has_feature("read"))
  self.assertTrue(self.adp.has_feature("read_from_file"))
  self.assertFalse(self.adp.has_feature("write"))
+ self.assertTrue(self.adp.has_feature("hooks"))
 
  def test_pass_arguments_to_adapter(self):
  self.assertEqual(self.adp.read_from_file("foo", suffix=3).name, "foo3")

tests/test_hooks_plugins.py

@@ -19,6 +19,7 @@
 
 HOOKSCRIPT_PATH = "hookscript_example"
 POST_WRITE_HOOKSCRIPT_PATH = "post_write_hookscript_example"
+CUSTOM_ADAPTER_HOOKSCRIPT_PATH = "custom_adapter_hookscript_example"
 
 POST_RUN_NAME = "hook ran and did stuff"
 TEST_METADATA = {'extra_data': True}
@@ -71,8 +72,17 @@ def setUp(self):
  "baselines",
  POST_WRITE_HOOKSCRIPT_PATH
  )
- self.man.hook_scripts = [self.hsf, self.post_hsf]
-
+ self.adapter_hook_jsn = baseline_reader.json_baseline_as_string(
+ CUSTOM_ADAPTER_HOOKSCRIPT_PATH
+ )
+ self.adapter_hookscript = otio.adapters.otio_json.read_from_string(
+ self.adapter_hook_jsn)
+ self.adapter_hookscript._json_path = os.path.join(
+ baseline_reader.MODPATH,
+ "baselines",
+ HOOKSCRIPT_PATH
+ )
+ self.man.hook_scripts = [self.hsf, self.post_hsf, self.adapter_hookscript]
  self.orig_manifest = otio.plugins.manifest._MANIFEST
  otio.plugins.manifest._MANIFEST = self.man
 
@@ -83,6 +93,8 @@ def tearDown(self):
  def test_plugin_adapter(self):
  self.assertEqual(self.hsf.name, "example hook")
  self.assertEqual(self.hsf.filepath, "example.py")
+ self.assertEqual(otio.adapters.from_name("example").adapter_hook_names(),
+ ["custom_adapter_hook"])
 
  def test_load_adapter_module(self):
  target = os.path.join(
@@ -101,15 +113,25 @@ def test_run_hook_function(self):
  self.assertEqual(result.name, POST_RUN_NAME)
  self.assertEqual(result.metadata.get("extra_data"), True)
 
+ def test_run_custom_hook_function(self):
+ tl = otio.schema.Timeline()
+ result = otio.hooks.run(hook="custom_adapter_hook", tl=tl,
+ extra_args=TEST_METADATA)
+ self.assertEqual(result.metadata["custom_hook"], TEST_METADATA)
+
  def test_run_hook_through_adapters(self):
+ hook_map = dict(TEST_METADATA)
+ hook_map["run_custom_hook"] = True
+
  result = otio.adapters.read_from_string(
  'foo', adapter_name='example',
  media_linker_name='example',
- hook_function_argument_map=TEST_METADATA
+ hook_function_argument_map=hook_map
  )
 
  self.assertEqual(result.name, POST_RUN_NAME)
  self.assertEqual(result.metadata.get("extra_data"), True)
+ self.assertEqual(result.metadata["custom_hook"]["extra_data"], True)
 
  def test_post_write_hook(self):
  self.man.adapters.extend(self.orig_manifest.adapters)
@@ -161,19 +183,20 @@ def test_available_hookscript_names(self):
  # for not just assert that it returns a non-empty list
  self.assertEqual(
  list(otio.hooks.available_hookscripts()),
- [self.hsf, self.post_hsf]
+ [self.hsf, self.post_hsf, self.adapter_hookscript]
  )
  self.assertEqual(
  otio.hooks.available_hookscript_names(),
- [self.hsf.name, self.post_hsf.name]
+ [self.hsf.name, self.post_hsf.name, self.adapter_hookscript.name]
  )
 
  def test_manifest_hooks(self):
  self.assertEqual(
  sorted(list(otio.hooks.names())),
  sorted(
  ["post_adapter_read", "post_media_linker",
- "pre_adapter_write", "post_adapter_write"]
+ "pre_adapter_write", "post_adapter_write",
+ "custom_adapter_hook"]
  )
  )
 
@@ -204,6 +227,13 @@ def test_manifest_hooks(self):
  ]
  )
 
+ self.assertEqual(
+ list(otio.hooks.scripts_attached_to("custom_adapter_hook")),
+ [
+ self.adapter_hookscript.name
+ ]
+ )
+
  tl = otio.schema.Timeline()
  result = otio.hooks.run("pre_adapter_write", tl, TEST_METADATA)
  self.assertEqual(result.name, POST_RUN_NAME)