4

Spoiler

My question is the following: Are there any design patterns for representing chainable functions that are for the problem described below?

High-Level Decription of the Process

I'm currently building an image-processing server, whose functionality is exposed via a web API.

Client Behavior

Clients first authenticate, then request that one or more analyses be performed by the server. Once a client has authenticated and requested his analysis stack, he streams his video data to the analysis server.

Server Behavior

On the server side, the Stream class is used to represent the flow of data. It's the first-class object around which the pipeline is built, and it can be thought of as an ordered FIFO queue of sorts. Frames are pushed, in order of arrival, onto the Stream instance, and each stream may optionally map a function onto its data. Results from a mapped function are projected onto a new Stream instance.

For example:

# input_stream is asynchronously updated with incoming frames

stream = input_stream.map(do_sobel_filter)  # apply Sobel filter to each incoming frame
stream2 = stream.window(fn, 10)  # apply function to a sliding window of 10 frames

An important side-note is that our use of Stream objects allows us to apply functions either to individual frames or to groups of frames. This is useful as some analyses require context.

Requirements & Constraints

The above bold text is particularly important; A given client will be authorized only to perform a subset of the image analyses offered by our system. This in turn makes it critical that we be able to dynamically compose the image-processing pipeline -- we don't want to perform expensive computations only to throw away the result.

Further complicating the matter is the fact that almost all of the analyses we offer have some prerequisites, that is, operation Oi may depend on the result of operation Oi-n. Our system needs to resolve these dependencies and run each operation exactly once.

Finally, the chaining/composition of operations must be associative. This is because we'd like to be able to compose reusable higher-level computations from lower-level atomic computations. For instance, consider the sequence of operations x -> y -> z. We would like to be able to assign the ordered chain x->y->z to a variable, thus CMP1 = x->y->z. Similarly, CMP2 is the result of chaining operations a->b->c. Ultimately, we would like to be able to do the equivalent of CMP3 = CMP1 -> CMP2, and have this be strictly equivalent to CMP3 = x->y->z->a->b->c (associativity).

To summarize, here are the requirements for my image analysis function objects:

  1. Functions must be composeable, meaning that it should be possible to manipulate a first-class object that represents a series of sub-computations.
  2. Ideally, functions should resolve dependencies, meaning that doing A -> B should implicitly evaluate to A -> X -> C if C depends on both A and X. This may be worthy of it's own question, and so should be treated as a soft requirement.
  3. Chaining of operations should be associative, and composed functions should also be chainable. The idea is to have a reusable pattern for representing increasingly complex pipelines.

Are there any well-tested patterns for achieving such a result?

Edits:

1) In response to @Giorgio's questions, my (admittedly ambiguous) notation serves to distinguish between the temporal order of frames and the order of operations. Below is an example of how an operation can depend on its preceding counterparts:

FindFace -> FindEyes -> SegmentEye -> MeasurePupilDiameter

Here we can see that each function after FindFace depends on the cumulative result of the preceding functions. Now consider:

FindFace -> FindEyes -> SegmentEye -> GetIrisColorHistogram

This second example demonstrates how there is often a common core of operations that needs to be performed across analyses. When composing operations, I'd like to do steps 1 - 3 only once, and then apply MeasurePupilDiameter and GetIrisColorHistogram from there.

Intuitively, it seems like the right way to do this would be to assign a dictionary to my Frame object and fill it with features as they're computed -- a memoize pattern, if you will. From there, however, it's unclear to me how I should go about abstracting all the boilerplate code pertaining to (a) checking if a prerequisite feature has already been computed and (b) automatically calling the appropriate function if it has not.

  • 2
    Maybe you can define a monad that can help you structure your solution: monads offer associative chaining of functions. – Giorgio Dec 14 '14 at 15:12
  • A few questions. (1) Are you applying a fixed operation f to each new image, or what does the function to be applied to each image depend on? (2) By operation Oi, do you mean the application of such a function f to image i? (3) When an operation Oi depends on the result of a previous operation O(i-n), how far in the past is this previous operation? (4) In general, can an operation depend on the result of one previous operation only, or on more than one (list of previous results)? – Giorgio Dec 14 '14 at 15:26
  • @Giorgio, I thought about this solution and did a bunch of reading ("state monad", in particular, got my attention), but I must admit I'm having trouble making the leap from tutorial to implementation. At the very least your comment is reassuring insofar as I'm on the right track -- If you have any specific ideas, I'd be very interested in hearing them. – blz Dec 14 '14 at 15:47
  • @Giorgio, to answer your questions: (1) Yes, I'm defining a fixed operation to each frame. This having been said, it may be relevant to modify the state of the function object based on the contents of a frame, such that subsequent frames will be handled differently. (2-3) My Oi notation is actually quite ambiguous -- I mean that "an operation", O, is preceded and succeeded by other operations. Thus, Oi-n is the operation that happens n operations before Oi. In general, operations can depend on the results of several previous ones (see edits). – blz Dec 14 '14 at 15:53
  • So, if I understand correctly, Oi indicate different operations on the same frame. So you never have an operation on frame j depend on a previous operation performed on an earlier frame, am I correct? – Giorgio Dec 14 '14 at 18:19
3
+50

Your problem has striking parallels to functional programming: Monads, Functors, and composing functions. Your composability requirements essentially states that each operation must be a function that takes a stream and returns a stream:

operation : Stream -> Stream

Most functions will not be expressed in terms of whole streams but rather single frames or windows of frames. You have correctly noted that you can lift these frame-wise operations to stream operations by functions like map.

class Operation:
    def __init__(self, function, dependencies):
        self.dependencies = dependencies
        self.function = function
    def __call__(self, stream):
        return self.function(stream)

def lift_frame_to_stream(frame_operation, dependencies):
    def stream_operation(stream):
        for frame in stream:
            yield frame_operation(frame)
    return Operation(stream_operation, dependencies)

This composability is at odds with the requirement that the operations can have dependencies. We can solve this by handling the dependencies outside of the type system. Given a flat list of operations op1, op2, ..., opn that each have dependencies, then we can iterate through the list to determine whether these dependencies are fulfilled. In the simplest case, this is done by keeping a set of all operations encountered up to that point:

# unsatisfied_dependencies: Iterable[Operation] -> List[(Operation, List[Operation])]
def unsatisfied_dependencies(operations):
    seen = set()
    needed = []
    for op in operations:
        op_needs = []
        for dep in op.dependencies:
            if dep not in seen:
                op_needs.append(dep)
        if op_needs:
            needed.append((op, op_needs))
        seen.add(op)
    return needed

I would be wary of automatically fulfilling dependencies. If this would be done, you'd have to iterate the operations by index. If an unmet dependency is discovered, rather than adding it to the list of needed operations you'd insert the dependency before the current operation, then back up one element, and continue checking the dependencies starting with the newly inserted operation.

The problem with automatic dependency insertion is that the order of operations could matter, and this resolution strategy does not account for that. I think it would be better to make users aware of dependencies and require these to be met before processing starts.

Processing the stream is a fairly simple job of taking a pipeline with all dependencies met and applying the operations. Assuming that all operations are lazy (i.e. don't immediately consume the stream but only calculate each frame that is requested; easily implemented using Python's generators), then the composed stream would have to be force-evaluated.

stream = input_stream
for op in verified_pipeline:
    stream = op(stream)
# force the stream

The pipeline cannot be composed any other way; “associativity” is a misnomer since stream operations are unary and not binary operators. However, your requirement for larger “building blocks“ of operations that contain child operations is understandable, and easy to implement – the object-oriented Composite Pattern makes sense here. The dependencies of the composite are all dependencies of the child operations that are not met inside that composite.

Actually, a pipeline of operations is equivalent to a single operation, so we can use a pipeline as the composite.

class Pipeline:
    def __init__(self, childs, extra_dependencies=[]):
        self.childs = childs
        deps = []
        for (op, dependencies) in unsatisfied_dependencies(childs):
            deps.extend(dependencies)
        deps.extend(extra_dependencies)
        self.dependencies = deps

    def __call__(self, stream):
        for op in self.childs:
            stream = op(stream)
        return stream

More important than the specifics of composing operations is how you'll represent each frame. Supposedly, each frame will consist primarily of image data. However, some operations might want to add metadata (e.g. one operation might calculate the colour distribution of a frame). It would therefore make a lot of sense to have each frame contain a dicitionary that can be filled with fairly unstructured data. Since we want the operations to be freely composable, it would not be viable to define multiple frame classes as output of each operation since that way metadata from previous operations would be discarded.

But this permanent metadata has an important problem: it can be invalidated by subsequent operations. E.g. a colour profile would be invalidated by an operation that performs a gamma correction on the image data. One possibility to reconcile this would be to track for each operation which operations it invalidates. When checking for unmet dependencies, such invalidated operations would then be removed from the seen state. However, this requires you to do extensive bookkeeping for each operation. Adding one new operation involves looking through all existing operations to discover potential incompatibilities.

Regarding first-class representation of the operations, I think that representing each operation as a callable object with some metadata such as dependencies tacked on to it makes most sense. As shown above, it's easy to verify dependencies and build composable blocks out of this. It is also fairly easy to turn user input into a pipeline. Assuming you get a list of strings that refer to some operation, you can use a dictionary to map these strings to operations:

operations_map = { ... }

chosen_operations = []
for name in input_list:
    chosen_operations.append(operations_map[name]) # TODO error handling
pipeline = Pipeline(chosen_operations)

if pipeline.dependencies:
    # oops

result_stream = pipeline(input_stream)

However, it might be good to have each operation contain a name field to ease creation of meaningful error messages etc.

  • I'm so glad you brought this up! Monads seemed to be the correct solution and I've been doing a lot of reading on the subject ... Your answer seems like the missing link between theory and implementation. I'll read it carefully and get back to you, thanks! – blz Dec 14 '14 at 16:07
  • @blz While this answer was inspired by monads and related constructs, it doesn't actually contain any monads itself. The suggested lift_frame_to_stream: (Frame -> Frame) -> (Stream -> Stream) function is however an example of fmap (map for Functors, which are a less expressive relative of Monads). – amon Dec 14 '14 at 16:09
  • duly noted! Thanks for the new Google search term :) – blz Dec 14 '14 at 16:12
  • re your argument against dynamically resolving dependencies during processing: do you think it would be acceptable to resolve dependencies at composition-time? That is, dependencies could be resolved while building the Stream objects and chaining them together, such that everything is in working order when the first Frame instance is passed through the Operation. This could presumably be done by passing a set between Streams when mapping -- a writer monad might be a good way of achieving this. You seem to have some experience in the matter, so I'd love to hear your thoughts. – blz Dec 14 '14 at 17:02
  • @blz: Monad could be useful if the operation you apply to element i depends on the result of the computation performed on element i - 1. If you are only mapping from individual stream elements, then it is just a functor map (fmap), as amon has pointed out. – Giorgio Dec 14 '14 at 18:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.