Execution Graph#

Directed acyclic graph (DAG) that composes pipeline modules, resolves dependencies, and drives execution in the correct order.

Builds an execution DAG from a list of modules.

The builder inspects each module’s declared inputs and outputs, then wires nodes together so that every node’s dependencies point to the nodes that produce its required inputs.

class adapt.execution.graph.builder.GraphBuilder(modules)#

Bases: object

Construct a directed acyclic graph from module declarations.

Parameters:

modules (list[BaseModule]) – Ordered or unordered list of modules to connect. Execution order is determined by input/output matching, not list order.

Raises:

  • ValueError – If two modules declare the same output key (ambiguous dependency).

  • Example:: – builder = GraphBuilder([acquisition_mod, ingest_mod, detection_mod, projection_mod]) nodes = builder.build()

build()#

Build and return the list of connected nodes.

Returns:

All nodes with dependencies and dependents populated. Nodes are returned in insertion order; execution order is determined by the GraphExecutor.

Return type:

list[Node]

Executes a DAG of nodes in dependency order.

The executor runs a simple topological loop: nodes whose dependencies have all completed are eligible to run. This single-threaded executor is the reference implementation. Future steps may add parallel execution.

Contract enforcement#

If a module declares input_contracts or output_contracts, the executor calls each validator automatically before and after run(). Validators raise ContractViolation on failure; the exception propagates up to the caller (e.g. RadarProcessor) which handles it.

class adapt.execution.graph.executor.GraphExecutor(nodes, observability=None)#

Bases: object

Execute a list of nodes in topological order.

Parameters:

  • nodes (list[Node]) – Connected nodes returned by GraphBuilder.build().

  • Example:: – nodes = GraphBuilder([a, b, c]).build() executor = GraphExecutor(nodes) result_context = executor.run(initial_context={})

  • observability (Observability | None)

run(context)#

Execute all nodes in dependency order.

Parameters:

context (dict) – Initial data available to the first nodes (e.g. file paths, config). Each node’s outputs are merged into this dict after the node runs, making them available to downstream nodes.

Returns:

Final context containing all module outputs.

Return type:

dict

Raises:

RuntimeError – If the graph contains a cycle (nodes that can never be ready).

Execution graph node.

Each node wraps a module and tracks its position in the DAG: which nodes it depends on (must complete before this node runs) and which nodes depend on it (notified when this node completes).

class adapt.execution.graph.node.Node(module)#

Bases: object

A node in the execution graph.

Parameters:

module (BaseModule) – The processing module this node wraps.

module#
Type:

BaseModule

inputs#

Data keys required by this module.

Type:

list[str]

outputs#

Data keys produced by this module.

Type:

list[str]

dependencies#

Upstream nodes that must complete before this one runs.

Type:

list[Node]

dependents#

Downstream nodes that are unblocked when this one completes.

Type:

list[Node]

property name: str#

Module registry — discover and instantiate registered modules.

Modules register their class at import time by calling registry.register(). The controller calls registry.create_modules() to get instantiated modules ready for graph construction.

Usage:

# In a module file (modules/detection/module.py):
from adapt.execution.module_registry import registry

class DetectModule(BaseModule):
    name = "detection"
    inputs = ["grid_volume"]
    outputs = ["storm_cells"]
    ...

registry.register(DetectModule)

# In the controller:
modules = registry.create_modules()
nodes = GraphBuilder(modules).build()
GraphExecutor(nodes).run(context)
class adapt.execution.module_registry.ModuleRegistry#

Bases: object

Central registry for Adapt modules and extensions.

Stores module classes. Instantiates them on demand via create_modules(). This allows each module to be configured with constructor arguments at runtime while keeping registration declarative.

Raises:

  • RuntimeError – If the same module name is registered twice.

  • KeyError – If get() is called with an unregistered name.

clear()#

Remove all registered modules (primarily for testing).

Return type:

None

create_modules()#

Instantiate and return all registered modules.

Returns:

One instance per registered module class, in registration order.

Return type:

list[BaseModule]

get(name)#

Return the module class registered under name.

Raises:

KeyError – If no module with that name is registered.

Return type:

type[BaseModule]

Parameters:

name (str)

list_modules()#

Return names of all registered modules.

Return type:

list[str]

register(module_class)#

Register a module class by its name attribute.

Parameters:

module_class (type[BaseModule]) – The module class to register. Must have a non-empty name.

Raises:

  • RuntimeError – If a module with the same name is already registered.

  • ValueError – If module_class.name is empty.

Return type:

None

unregister(name)#

Remove a module from the registry (primarily for testing).

Return type:

None

Parameters:

name (str)