Il tuo problema ha sorprendenti analogie con la programmazione funzionale: Monade, Funzionalità e funzioni di composizione. I requisiti di composabilità stabiliscono essenzialmente che ciascuna operazione deve essere una funzione che accetta uno stream e restituisce uno stream:

operation : Stream -> Stream

La maggior parte delle funzioni non saranno espresse in termini di interi flussi ma piuttosto di singoli fotogrammi o finestre di fotogrammi. Hai correttamente notato che puoi sollevare queste operazioni frame-wise per eseguire operazioni di streaming con funzioni come 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)

Questa componibilità è in contrasto con il requisito che le operazioni possano avere dipendenze. Possiamo risolvere questo problema gestendo le dipendenze al di fuori del sistema di tipi. Dato un elenco semplice di operazioni op1, op2, ..., opn ognuna delle quali ha dipendenze, allora possiamo scorrere l'elenco per determinare se queste dipendenze sono soddisfatte. Nel caso più semplice, ciò avviene mantenendo un insieme di tutte le operazioni incontrate fino a quel punto:

# 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

Sarei diffidente nei confronti delle dipendenze che soddisfano automaticamente. Se questo fosse fatto, dovresti iterare le operazioni per indice. Se viene rilevata una dipendenza non soddisfatta, anziché aggiungerla all'elenco delle operazioni necessarie, inserire la dipendenza prima dell'operazione corrente, quindi eseguire il backup di un elemento e continuare a controllare le dipendenze a partire dall'operazione appena inserita.

Il problema con l'inserimento automatico delle dipendenze è che l'ordine delle operazioni potrebbe avere importanza e questa strategia di risoluzione non tiene conto di ciò. Penso che sarebbe meglio rendere gli utenti consapevoli delle dipendenze e richiedere che vengano soddisfatti prima dell'inizio dell'elaborazione.

L'elaborazione del flusso è un compito abbastanza semplice di prendere una pipeline con tutte le dipendenze soddisfatte e applicando le operazioni. Supponendo che tutte le operazioni siano pigre (cioè non consumano immediatamente il flusso ma calcolano solo ogni frame richiesto, facilmente implementabile usando i generatori di Python), il flusso composto dovrebbe essere sottoposto a valutazione forzata.

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

La pipeline non può essere composta in nessun altro modo; "Associatività" è un termine improprio poiché le operazioni di flusso sono unate e non operatori binari. Tuttavia, la tua esigenza di "blocchi" di operazioni più grandi che contengono operazioni secondarie è comprensibile e facile da implementare - il Pattern composito orientato agli oggetti ha senso qui. Le dipendenze del composito sono tutte le dipendenze delle operazioni secondarie che non sono soddisfatte all'interno di quel composito.

In realtà, una pipeline di operazioni è equivalente a una singola operazione, quindi possiamo usare una pipeline come composito.

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

Più importante delle specificità delle operazioni di composizione è il modo in cui rappresenterai ciascun fotogramma. Presumibilmente, ogni frame sarà costituito principalmente da dati immagine. Tuttavia, alcune operazioni potrebbero voler aggiungere metadati (ad esempio un'operazione potrebbe calcolare la distribuzione del colore di un frame). Sarebbe quindi molto sensato che ogni frame contenga un dicitionary che può essere riempito con dati abbastanza non strutturati. Dal momento che vogliamo che le operazioni siano liberamente componibili, non sarebbe possibile definire più classi di frame come output di ogni operazione, poiché in questo modo i metadati delle operazioni precedenti verrebbero scartati.

Ma questi metadati permanenti hanno un problema importante: possono essere invalidati dalle operazioni successive. Per esempio. un profilo colore verrebbe invalidato da un'operazione che esegue una correzione gamma sui dati dell'immagine. Una possibilità per riconciliare questo sarebbe tracciare per ogni operazione quali operazioni invalida. Quando si controllano le dipendenze non soddisfatte, tali operazioni invalidate verrebbero quindi rimosse dallo stato seen . Tuttavia, questo richiede di fare una contabilità estesa per ogni operazione. L'aggiunta di una nuova operazione comporta la ricerca di tutte le operazioni esistenti per scoprire potenziali incompatibilità.

Per quanto riguarda la rappresentazione di prima classe delle operazioni, penso che rappresentare ogni operazione come un oggetto callable con alcuni metadati come le dipendenze su di esso abbia più senso. Come mostrato sopra, è facile verificare le dipendenze e creare blocchi componibili al di fuori di questo. È anche abbastanza semplice trasformare l'input dell'utente in una pipeline. Supponendo di avere un elenco di stringhe che fanno riferimento ad alcune operazioni, puoi usare un dizionario per mappare queste stringhe alle operazioni:

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)

Tuttavia, potrebbe essere utile che ogni operazione contenga un campo name per facilitare la creazione di messaggi di errore significativi ecc.