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.