The @edge Decorator — Under the Hood

What @edge Does

The @edge decorator is the primary entry point for compiling Python functions into native machine code within the Pyvorin Edge SDK. When you decorate a function with @edge, the Pyvorin compiler intercepts the function definition, parses its Abstract Syntax Tree (AST), infers types from annotations, and routes the compilation to either the scalar compiler (PyvorinCompiler) or the kernel backend (CKernelBackend) depending on the workload characteristics.

Under the hood, @edge does not merely wrap your function — it triggers a full compilation pipeline that produces a position-independent .so shared object, generates an ABI contract, and replaces the Python callable with a thin ctypes wrapper that invokes the native code. This process is lazy: compilation happens on first call or when explicitly requested via CompilerBridge.compile_hotpath().

AST Parsing and Analysis

The compiler bridge uses Python's built-in ast module to parse the decorated function's source code. The AST walk serves three critical purposes:

• Workload detection: The bridge distinguishes between scalar workloads (simple arithmetic, conditionals) and kernel workloads (loops over arrays, reductions like sum, min, max).
• Type inference: When type hints are present, the bridge maps Python types to ctypes equivalents (int → c_int64, float → c_double, bool → c_bool, str → c_char_p).
• Operation extraction: For kernel workloads, the bridge extracts a list of KernelOp objects describing arithmetic, comparisons, calls, loops, and conditionals.

Here is the actual heuristic used by CompilerBridge._detect_workload_type():

from pyvorin_edge.compiler_bridge import CompilerBridge

source = '''
def moving_average(readings: list[float]) -> float:
    total = 0.0
    for r in readings:
        total = total + r
    return total / len(readings)
'''

workload = CompilerBridge._detect_workload_type(source, "moving_average")
print(workload)  # "kernel" because it contains a For loop and array operations
  

The AST walker looks for ast.For, ast.While, ast.ListComp, ast.GeneratorExp, and calls to reduction builtins (sum, min, max, mean, std, var). If both loops and array operations are present, the workload is classified as "kernel"; otherwise it is "scalar".

Type Inference from Source

When type hints are absent, the compiler falls back to a conservative [c_int64, c_int64] → c_int64 signature. With hints, the bridge produces a precise ctypes signature:

source = '''
def threshold_alert(value: float, limit: float) -> bool:
    return value > limit
'''

arg_types, return_type = CompilerBridge._infer_signature_from_source(
    source, "threshold_alert"
)
print(arg_types)    # [, ]
print(return_type)  # 
  

The type map is intentionally narrow: only int, float, bool, and str are supported. Complex types, generics, and custom classes will silently fall back to c_int64, which often leads to runtime errors if the native code receives an incompatible pointer.

Compilation Routing

Once the workload type and signature are known, compile_hotpath() routes to the appropriate backend:

• Scalar backend (PyvorinCompiler): Handles straight-line arithmetic, conditionals, and small loops. Produces a .so via the main compiler pipeline.
• Kernel backend (CKernelBackend): Handles array traversals and reductions. Constructs a TypedKernelPlan with ArrayArg and ScalarArg descriptors, then compiles to a .so via the C kernel backend.

Kernel compilation is best-effort. If the kernel backend fails, the bridge logs a warning and falls back to scalar compilation. If scalar compilation also fails, a CompilerBridgeError is raised.

Full Annotated Example

from pathlib import Path
from pyvorin_edge.compiler_bridge import CompilerBridge

bridge = CompilerBridge()

# 1. Define a hot-path function as a string (in practice, @edge reads __source__)
source_code = '''
def compute_power(voltage: float, current: float) -> float:
    return voltage * current
'''

# 2. Compile to a .so
output = Path("/tmp/edge_modules/power.so")
so_path = bridge.compile_hotpath(source_code, "compute_power", output)
print(f"Compiled module: {so_path}")

# 3. Generate an ABI contract for the .so
abi = bridge.generate_abi(so_path, "compute_power")
print(abi.to_dict())

# 4. Validate the compiled code against the reference Python function
def reference_compute_power(voltage: float, current: float) -> float:
    return voltage * current

report = bridge.validate_compilation(
    so_path,
    reference_compute_power,
    test_inputs=[(230.0, 10.0), (120.0, 5.0), (0.0, 0.0)],
)
print(report)
# {'passed': 3, 'failed': 0, 'mismatches': []}
  

When to Use @edge

• Rule conditions: Pipeline rule expressions that evaluate thousands of times per second.
• Window aggregations: Rolling mean, min, max over sensor buffers.
• Math kernels: Vector dot products, RMS calculations, threshold comparisons.
• Battery-powered devices: Reducing CPU time directly translates to longer battery life on Raspberry Pi and embedded Linux boards.

When NOT to Use @edge

• I/O-bound code: If your function spends most of its time in read(), write(), or network calls, native compilation will not help.
• Dynamic types: Functions that accept Any, unions, or mixed-type collections cannot be safely compiled.
• Recursion: The compiler does not optimize recursive calls; they may stack-overflow or fail to compile entirely.
• Eval / exec / getattr: Any meta-programming that inspects or modifies code at runtime is incompatible with ahead-of-time compilation.
• Large standard-library usage: Only a small safe subset of builtins (abs, min, max, sum, len, round, float, int, bool) are available in compiled rules.

Known Limitations

Pipeline Integration

The compiler bridge can compile an entire pipeline's rule conditions in one pass:

from pyvorin_edge.pipeline import Pipeline, RuleConfig
from pyvorin_edge.compiler_bridge import CompilerBridge

pipeline = Pipeline(name="hvac_monitor")
pipeline.add_rule(
    RuleConfig(
        name="overheat",
        condition_expr="ctx.value > 28.0",
        severity="critical",
    )
)

bridge = CompilerBridge()
compiled = bridge.compile_pipeline(pipeline, output_dir="/tmp/edge_modules/")
manifest = bridge.generate_manifest(pipeline, compiled)
print(json.dumps(manifest, indent=2))
  

This produces a manifest.json describing every compiled module, its path, and its ABI contract, which the edge runtime loads at startup to replace interpreted rule expressions with native equivalents.