Last active
December 13, 2017 17:54
-
-
Save tokejepsen/acd7d86f268e579883c8da0b1bb7d9ac to your computer and use it in GitHub Desktop.
Flow Based Programming Experiments
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
- Keyword arguments can be None, so need something more unique to compare | |
against. | |
- There is the possibility of cache the results of each component for quicker | |
evaluation. | |
""" | |
import inspect | |
import functools | |
from pyblish import api | |
class Component(object): | |
def __init__(self, method): | |
self.method = method | |
# Class methods validation | |
if hasattr(method, "im_self"): | |
msg = "Class methods needs to be bound." | |
assert method.im_self is not None, msg | |
# Get arguments | |
args = inspect.getargspec(self.method).args | |
kwargs = get_keyword_arguments(method).keys() | |
arguments = [x for x in args if x not in kwargs] | |
# Assign arguments | |
self.arguments = {} | |
for arg in arguments: | |
# Ignore "self" argument of class methods | |
if arg == "self": | |
continue | |
self.arguments[arg] = None | |
# Assign keyword arguments | |
self.keyword_arguments = {} | |
for arg in get_keyword_arguments(self.method).keys(): | |
self.keyword_arguments[arg] = None | |
def __str__(self): | |
return "<class Component '{0}'>".format(self.method) | |
def process(self): | |
arguments = [] | |
for key, value in self.arguments.iteritems(): | |
if value is None: | |
raise ValueError( | |
"Argument \"{0}\" could not be resolved on Component " | |
"{1}".format(key, self) | |
) | |
else: | |
arguments.append(value()) | |
keyword_arguments = get_keyword_arguments(self.method) | |
for key, value in self.keyword_arguments.iteritems(): | |
if value is None: | |
self.keyword_arguments[key] = keyword_arguments[key] | |
if callable(value): | |
self.keyword_arguments[key] = value() | |
return self.method(*arguments, **self.keyword_arguments) | |
def get_keyword_arguments(method): | |
arguments = inspect.getargspec(method).args | |
defaults = inspect.getargspec(method).defaults | |
keyword_arguments = {} | |
if defaults: | |
for arg in arguments[-len(defaults):]: | |
keyword_arguments[arg] = defaults[len(keyword_arguments.keys())] | |
return keyword_arguments | |
# Assign "process" method to Components values | |
def resolve_connections(connections): | |
for connection in connections: | |
source = connection[0] | |
target = connection[1] | |
arg = connection[2] | |
msg = "{0} is not a Component in connection {1}" | |
assert isinstance(source, Component), msg.format(source, connection) | |
assert isinstance(target, Component), msg.format(target, connection) | |
# Resolve arguments | |
if arg in target.arguments.keys(): | |
target.arguments[arg] = functools.partial(source.process) | |
# Resolve keyword arguments | |
if arg in target.keyword_arguments.keys(): | |
target.keyword_arguments[arg] = functools.partial(source.process) | |
# Resolve *args | |
inspection = inspect.getargspec(target.method) | |
if inspection.varargs and arg == inspection.varargs: | |
num_args = len(target.arguments.keys()) | |
arg = "{0}{1}".format(arg, num_args + 1) | |
target.arguments[arg] = functools.partial(source.process) | |
# Resolve **kwargs | |
if inspection.keywords and arg == inspection.keywords: | |
num_args = len(target.keyword_arguments.keys()) | |
arg = "{0}{1}".format(arg, num_args + 1) | |
target.keyword_arguments[arg] = functools.partial(source.process) | |
# Test connection between Component with/without arguments | |
def add(x, y): | |
return x + y | |
def x(): | |
return 1 | |
x1 = Component(x) | |
add1 = Component(add) | |
connections = [ | |
(x1, add1, "x"), | |
(x1, add1, "y"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 2 | |
# Test nested connections | |
add2 = Component(add) | |
connections = [ | |
(add1, add2, "x"), | |
(add1, add2, "y"), | |
] | |
resolve_connections(connections) | |
assert add2.process() == 4 | |
# Test Components with single keyword method | |
def add(x, y=1): | |
return x + y | |
add1 = Component(add) | |
connections = [ | |
(x1, add1, "x"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 2 | |
# Test Components with multiple keyword method | |
def add(x=2, y=1): | |
return x + y | |
add1 = Component(add) | |
assert add1.process() == 3 | |
# Test Components with overriding keyword arguments | |
add1 = Component(add) | |
connections = [ | |
(x1, add1, "x"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 2 | |
# Test overriding keyword argument. | |
def y(): | |
return 2 | |
y1 = Component(y) | |
connections = [ | |
(x1, add1, "x"), | |
(y1, add1, "y"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 3 | |
# Test class method Component. | |
class add(object): | |
y = 1 | |
def adding(self, x): | |
return x + self.y | |
add1 = Component(add().adding) | |
connections = [ | |
(x1, add1, "x") | |
] | |
resolve_connections(connections) | |
assert add1.process() == 2 | |
# Test class attribute overriding | |
add_cls = add() | |
add_cls.y = 2 | |
add1 = Component(add_cls.adding) | |
connections = [ | |
(x1, add1, "x") | |
] | |
resolve_connections(connections) | |
assert add1.process() == 3 | |
# Test supporting *args | |
def add(*args): | |
return sum(args) | |
add1 = Component(add) | |
connections = [ | |
(x1, add1, "args"), | |
(x1, add1, "args"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 2 | |
# Test supporting **kwargs | |
def add(**kwargs): | |
return sum(kwargs.values()) | |
add1 = Component(add) | |
connections = [ | |
(x1, add1, "kwargs"), | |
(x1, add1, "kwargs"), | |
(x1, add1, "kwargs"), | |
] | |
resolve_connections(connections) | |
assert add1.process() == 3 | |
# Proof of concept for Pyblish | |
""" | |
+---------------+ +----------------+ | |
| CollectorA +--> CollectorA1 +-----+ | |
| ContextPlugin | | InstancePlugin | | | |
+---------------+ +----------------+ +v------------------+ +----------------+ | |
| collection +--> ValidatorA | | |
| Pause and inspect | | InstancePlugin | | |
+^------------------+ +----------------+ | |
+---------------+ | | |
| CollectorB +-------------------------+ | |
| ContextPlugin | | |
+---------------+ | |
""" | |
class CollectorA(api.ContextPlugin): | |
def process(self, context): | |
# We have to return something, but this could be handle with a wrapper | |
# of some kind instead. | |
print "Processing CollectorA" | |
context.create_instance(name="InstanceA1") | |
context.create_instance(name="InstanceA2") | |
return context | |
class CollectorA1(api.InstancePlugin): | |
def return_context(self, context): | |
# Since we have passing the context along the chain, we need to input | |
# and return a context. This could probably also be handled with a | |
# wrapper. | |
print "Processing CollectorA1" | |
for instance in context: | |
self.process(instance) | |
return context | |
def process(self, instance): | |
print instance.data["name"] | |
class CollectorB(api.ContextPlugin): | |
def process(self, context): | |
print "Processing CollectorB" | |
context.create_instance(name="InstanceB1") | |
context.create_instance(name="InstanceB2") | |
return context | |
class ValidatorA(api.InstancePlugin): | |
def return_context(self, context): | |
print "Processing ValidatorA" | |
for instance in context: | |
self.process(instance) | |
return context | |
def process(self, instance): | |
print instance.data["name"] | |
def context(): | |
"""Get a context.""" | |
return api.Context() | |
def collection(*contexts): | |
"""Merging all incoming contexts.""" | |
context = api.Context() | |
for cxt in contexts: | |
context.data.update(cxt.data) | |
for instance in cxt: | |
context.add(instance) | |
print "context: {0}".format(context) | |
raw_input("Press any key to continue to validation...") | |
return context | |
# Define all the components | |
context_component = Component(context) | |
CollectorA_component = Component(CollectorA().process) | |
CollectorA1_component = Component(CollectorA1().return_context) | |
CollectorB_component = Component(CollectorB().process) | |
collection_component = Component(collection) | |
ValidatorA_component = Component(ValidatorA().return_context) | |
# Define all the connections | |
connections = [ | |
(context_component, CollectorA_component, "context"), | |
(CollectorA_component, CollectorA1_component, "context"), | |
(CollectorA1_component, collection_component, "contexts"), | |
(context_component, CollectorB_component, "context"), | |
(CollectorB_component, collection_component, "contexts"), | |
(collection_component, ValidatorA_component, "context"), | |
] | |
resolve_connections(connections) | |
ValidatorA_component.process() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment