Lambda Executor

Lambda Executor: Python-Powered Custom Automation

The Lambda Executor enables you to run custom Python scripts as part of your automation workflows, providing unlimited flexibility for complex data processing, custom algorithms, external integrations, and advanced automation logic that goes beyond built-in features.

Info: This is an Executor node type that can be executed to perform work by the execution of a parent node. When triggered, it executes your Python script in a controlled environment.

Info: This feature uses TargetingNodeMetaData, which means it can be configured to read data from source nodes (Data Points) and write results to target nodes. Your Python script has access to source data and can update targets.

Warning: Lambda execution uses Python sandboxing for security. While powerful, ensure your scripts are tested and validated before deployment in production environments.

Overview

Lambda Executors bring the full power of Python to your automation workflows. Write custom scripts that can process data, interact with external APIs, perform complex calculations, control hardware, or implement any logic that your specific use case requires.

Key Features

• Python 3 Support: Write scripts in modern Python
• Source Data Access: Read from configured source data points
• Target Updates: Write results to target data points
• Custom Libraries: Use Python standard library
• Sandboxed Execution: Isolated execution environment for security
• Error Handling: Comprehensive error capture and logging
• Async Support: Non-blocking execution
• Script Validation: Pre-execution validation

Lambda Execution Flow

graph TD
    A[Parent Node Triggers] --> B[Load Python Script]
    B --> C[Fetch Source Data]
    C --> D[Prepare Execution Context]
    D --> E[Execute Python Script]
    E --> F{Execution Success?}
    F -->|Yes| G[Capture Script Output]
    F -->|No| H[Log Error]
    G --> I[Update Target Data Points]
    I --> J[Execute Children]
    H --> K[Set Error State]

How It Works

When a Lambda Executor runs:

1. Trigger: Parent node activates the Lambda executor
2. Script Loading: Retrieves the configured Python script
3. Data Fetching: Reads current values from source data points
4. Context Setup: Prepares execution environment with source data
5. Execution: Runs Python script in sandboxed environment
6. Result Capture: Captures script output/return value
7. Target Update: Updates target data points with results
8. Child Execution: Triggers any child executors if configured

Script Structure

Lambda scripts have access to:

• sources: Dictionary of source data point values
• print(): Output for logging
• Return value: Value to write to target data point

Basic script template:

# Access source data
temperature = sources.get('temperature', 0)
humidity = sources.get('humidity', 0)

# Perform calculations
heat_index = calculate_heat_index(temperature, humidity)

# Return result (written to target)
return heat_index

Configuration

Field	Description	Required
script	Python script code	Yes
sources	List of source data point IDs	No
targets	List of target data point IDs	No

Use Cases

• Complex Calculations: Algorithms beyond simple formulas
• Data Validation: Validate sensor data before processing
• Protocol Decoding: Parse custom data formats from serial devices
• External APIs: Call REST APIs with custom logic
• State Machines: Implement complex state logic
• Data Transformation: Convert between data formats
• Custom Algorithms: Implement proprietary calculations
• Serial Communication: Advanced serial device protocols

Example Scripts

Heat Index Calculation:

# Calculate heat index from temperature and humidity
temp = sources.get('temperature', 0)
humidity = sources.get('humidity', 0)

# Steadman heat index formula
hi = -42.379 + 2.04901523*temp + 10.14333127*humidity
hi -= 0.22475541*temp*humidity - 0.00683783*temp*temp
hi -= 0.05481717*humidity*humidity
hi += 0.00122874*temp*temp*humidity
hi += 0.00085282*temp*humidity*humidity
hi -= 0.00000199*temp*temp*humidity*humidity

return round(hi, 2)

Moving Average Filter:

# Keep last N readings and compute average
import json

# Load historical data from previous run
history = json.loads(sources.get('history', '[]'))
current = sources.get('sensor_value', 0)

# Add current value
history.append(current)

# Keep only last 10 readings
if len(history) > 10:
    history = history[-10:]

# Compute average
average = sum(history) / len(history)

# Store updated history for next run
print(f"History: {json.dumps(history)}")

return average

Data Validation:

# Validate sensor reading is within expected range
value = sources.get('sensor', 0)
min_valid = 0
max_valid = 100

if min_valid <= value <= max_valid:
    # Valid reading
    return value
else:
    # Invalid reading - return previous valid value
    print(f"Invalid reading: {value}")
    return sources.get('last_valid', 0)

Serial Data Parsing:

# Parse custom protocol from serial device
raw_data = sources.get('serial_data', '')

# Parse format: "TEMP:25.5,HUM:60.2,PRES:1013.2"
data = {}
for item in raw_data.split(','):
    key, value = item.split(':')
    data[key] = float(value)

# Return specific value
return data.get('TEMP', 0)

Python Standard Library Access

Lambda scripts can use Python standard library modules:

• math - Mathematical functions
• json - JSON parsing and encoding
• datetime - Date and time operations
• re - Regular expressions
• statistics - Statistical functions

Example:

import math
import statistics

values = [sources.get(f'sensor{i}', 0) for i in range(5)]
mean = statistics.mean(values)
stdev = statistics.stdev(values)

return mean

Security Considerations

The Lambda executor runs scripts in a sandboxed environment:

• Limited file system access
• Restricted network access (controlled by server configuration)
• Resource limits (CPU, memory)
• No access to system commands

Note: While sandboxing provides security, always review and test scripts before deployment, especially in production environments.

Error Handling

Lambda scripts should include error handling:

try:
    value = sources.get('sensor', 0)
    result = complex_calculation(value)
    return result
except Exception as e:
    print(f"Error: {e}")
    return 0  # Safe default

Integration Points

• Data Points: Read sensor values, write processed results
• Triggers: Execute scripts on schedules or conditions
• Executors: Chain with calculations, webhooks, logic gates
• Serial Devices: Process data from custom hardware
• MQTT: Process messages from MQTT topics

Best Practices

• Keep Scripts Simple: Focus on one task per script
• Error Handling: Always include try/except blocks
• Logging: Use print() for debugging and monitoring
• Performance: Avoid long-running operations
• Testing: Test scripts thoroughly before production use
• Documentation: Comment your code for future reference
• Validation: Validate input data before processing
• Default Values: Always provide safe defaults

Example Workflows

Custom Sensor Calibration:

1. Trigger: Cron Timer (every minute)
2. Executor: Lambda (apply calibration formula)
3. Target: Calibrated data point

Multi-Sensor Fusion:

1. Trigger: Data Point Update
2. Executor: Lambda (weighted average from multiple sensors)
3. Target: Fused sensor reading

Protocol Translator:

1. Trigger: Serial Device Data
2. Executor: Lambda (parse custom protocol)
3. Executor: Multiple data points (temperature, humidity, pressure)

Lambda Executors unlock unlimited automation possibilities by bringing the full power of Python programming to the Krill Platform.