Skip to content
Permalink
Browse files

Update the adapter.py to pass Argument Maps to Hooks (#450)

* Pass argument_maps through to hooks
  • Loading branch information...
mikekoetter authored and ssteinbach committed Mar 12, 2019
1 parent 0d6c8ce commit a216cc462d99a3c7765063eaa3ab18f774639a53
@@ -6,12 +6,12 @@ To write a new hook script, you create a python source file that defines a
a function named ``hook_function`` with signature:
``hook_function :: otio.schema.Timeline, Dict => otio.schema.Timeline``

The first argument is the timeline to process, and the second one is a dictionary of arguments that can be passed to it. Only one hook function can be defined per python file.
The first argument is the timeline to process, and the second one is a dictionary of arguments that can be passed to it by the adapter or media linker. Only one hook function can be defined per python file.

For example:
```python
def hook_function(tl, arg_dict):
def hook_function(tl, argument_map=None):
for cl in tl.each_clip():
cl.metadata['example_hook_function_was_here'] = True
return tl
@@ -79,3 +79,19 @@ To delete a function the list:
```
del hook_list[1]
```

## Example Hooks

### Replacing part of a path for drive mapping

An example use-case would be to create a pre-write adapter hook that checks the argument map for style being identified as nucoda and then preforms a path replacement on the reference url

```python
def hook_function(in_timeline,argument_map=None):
adapter_args = argument_map.get('adapter_arguments')
if if adapter_args and adapter_args.get('style') == 'nucoda':
for in_clip in in_timeline.each_clip():
''' Change the Path to use windows drive letters ( Nucoda is not otherwise forward slash sensative ) '''
if in_clip.media_reference:
in_clip.media_reference.target_url = in_clip.media_reference.target_url.replace(r'/linux/media/path','S:')
```
@@ -104,6 +104,7 @@ def read_from_file(
filepath,
media_linker_name=media_linker.MediaLinkingPolicy.ForceDefaultLinker,
media_linker_argument_map=None,
hook_function_argument_map={},
**adapter_argument_map
):
"""Execute the read_from_file function on this adapter.
@@ -135,8 +136,11 @@ def read_from_file(
**adapter_argument_map
)

# @TODO: pass arguments through?
result = hooks.run("post_adapter_read", result)
hook_function_argument_map['adapter_arguments'] = adapter_argument_map
hook_function_argument_map['media_linker_argument_map'] = \
media_linker_argument_map
result = hooks.run("post_adapter_read", result,
extra_args=hook_function_argument_map)

if media_linker_name and (
media_linker_name != media_linker.MediaLinkingPolicy.DoNotLinkMedia
@@ -147,20 +151,26 @@ def read_from_file(
media_linker_argument_map
)

# @TODO: pass arguments through?
result = hooks.run("post_media_linker", result)
result = hooks.run("post_media_linker", result,
extra_args=media_linker_argument_map)

return result

def write_to_file(self, input_otio, filepath, **adapter_argument_map):
def write_to_file(
self,
input_otio,
filepath,
hook_function_argument_map={},
**adapter_argument_map
):
"""Execute the write_to_file function on this adapter.
If write_to_string exists, but not write_to_file, execute that with
a trivial file object wrapper.
"""

# @TODO: pass arguments through?
input_otio = hooks.run("pre_adapter_write", input_otio)
hook_function_argument_map['adapter_arguments'] = adapter_argument_map
input_otio = hooks.run("pre_adapter_write", input_otio,
extra_args=hook_function_argument_map)

if (
not self.has_feature("write_to_file") and
@@ -183,6 +193,7 @@ def read_from_string(
input_str,
media_linker_name=media_linker.MediaLinkingPolicy.ForceDefaultLinker,
media_linker_argument_map=None,
hook_function_argument_map={},
**adapter_argument_map
):
"""Call the read_from_string function on this adapter."""
@@ -192,9 +203,12 @@ def read_from_string(
input_str=input_str,
**adapter_argument_map
)
hook_function_argument_map['adapter_arguments'] = adapter_argument_map
hook_function_argument_map['media_linker_argument_map'] = \
media_linker_argument_map

# @TODO: pass arguments through?
result = hooks.run("post_adapter_read", result)
result = hooks.run("post_adapter_read", result,
extra_args=hook_function_argument_map)

if media_linker_name and (
media_linker_name != media_linker.MediaLinkingPolicy.DoNotLinkMedia
@@ -205,17 +219,23 @@ def read_from_string(
media_linker_argument_map
)

# @TODO: pass arguments through?
# @TODO: Should this run *ONLY* if the media linker ran?
result = hooks.run("post_media_linker", result)
result = hooks.run("post_media_linker", result,
extra_args=hook_function_argument_map)

return result

def write_to_string(self, input_otio, **adapter_argument_map):
def write_to_string(
self,
input_otio,
hook_function_argument_map={},
**adapter_argument_map
):
"""Call the write_to_string function on this adapter."""

# @TODO: pass arguments through?
input_otio = hooks.run("pre_adapter_write", input_otio)
hook_function_argument_map['adapter_arguments'] = adapter_argument_map
input_otio = hooks.run("pre_adapter_write", input_otio,
extra_args=hook_function_argument_map)

return self._execute_function(
"write_to_string",
@@ -17,6 +17,7 @@
],
"hooks" : {
"pre_adapter_write" : ["example hook", "example hook"],
"post_adapter_read" : []
"post_adapter_read" : [],
"post_media_linker" : ["example hook"]
}
}
@@ -106,6 +106,15 @@ def test_run_hook_function(self):
self.assertEqual(result.name, POST_RUN_NAME)
self.assertEqual(result.metadata.get("extra_data"), True)

def test_run_hook_through_adapters(self):
result = otio.adapters.read_from_string('foo', adapter_name='example',
media_linker_name='example',
hook_function_argument_map=TEST_METADATA
)

self.assertEqual(result.name, POST_RUN_NAME)
self.assertEqual(result.metadata.get("extra_data"), True)

def test_serialize(self):

self.assertEqual(
@@ -143,7 +152,7 @@ def test_available_hookscript_names(self):
def test_manifest_hooks(self):
self.assertEqual(
sorted(list(otio.hooks.names())),
sorted(["post_adapter_read", "pre_adapter_write"])
sorted(["post_adapter_read", "post_media_linker", "pre_adapter_write"])
)

self.assertEqual(
@@ -159,6 +168,13 @@ def test_manifest_hooks(self):
[]
)

self.assertEqual(
otio.hooks.scripts_attached_to("post_media_linker"),
[
self.hsf.name
]
)

tl = otio.schema.Timeline()
result = otio.hooks.run("pre_adapter_write", tl, TEST_METADATA)
self.assertEqual(result.name, POST_RUN_NAME)

0 comments on commit a216cc4

Please sign in to comment.
You can’t perform that action at this time.