python

Overview

The python node executes Python code within your EdgeFlow pipeline. It injects the incoming message as a JSON object, runs your script via a wrapper, and captures the printed output back into the message. Supports custom Python paths and virtualenv isolation for dependency management — ideal for NumPy, pandas, scikit-learn, and other data-science workflows.

Properties

How It Works

import json, sys

# Automatically injected by EdgeFlow
msg = json.loads(sys.argv[1])

# Access message data
temp = msg["payload"]["temperature"]

# Your processing logic
result = temp * 9/5 + 32

# Print JSON to return data
print(json.dumps({
  "fahrenheit": result
}))

// Node configuration
useVirtualEnv: true
venvPath: "/home/pi/ml-env"

// EdgeFlow activates the venv, then runs:
// /home/pi/ml-env/bin/python script.py

// If script fails:
// msg.error.message = stderr text
// msg.error.source = "python"
// msg.error.code = exit code

// Use a catch node downstream

Example: NumPy Data Processing

Compute a moving average and standard deviation over a batch of sensor readings using NumPy.

// Python node configuration
{
  "code": "import json, sys, numpy as np\nmsg = json.loads(sys.argv[1])\nreadings = np.array(msg['payload']['values'])\nwindow = 5\nma = np.convolve(readings, np.ones(window)/window, mode='valid')\nprint(json.dumps({'moving_avg': ma.tolist(), 'std': float(np.std(readings)), 'mean': float(np.mean(readings))}))",
  "pythonPath": "python3",
  "useVirtualEnv": true,
  "venvPath": "/home/pi/data-env",
  "timeout": 30
}

// Input message
{
  "payload": {
    "values": [22.1, 22.4, 23.0, 22.8, 23.5, 24.1, 23.9, 24.5, 24.2, 25.0]
  }
}

// Output payload
{
  "moving_avg": [22.76, 23.16, 23.46, 23.76, 24.0, 24.34],
  "std": 0.89,
  "mean": 23.55
}

Example: ML Inference with scikit-learn

Load a pre-trained model and classify incoming sensor data at the edge.

// Python node configuration
{
  "code": "import json, sys, joblib\nimport numpy as np\nmsg = json.loads(sys.argv[1])\nmodel = joblib.load('/home/pi/models/anomaly_detector.pkl')\nfeatures = np.array([[msg['payload']['temperature'], msg['payload']['vibration'], msg['payload']['pressure']]])\nprediction = model.predict(features)[0]\nproba = model.predict_proba(features)[0]\nprint(json.dumps({'class': int(prediction), 'label': 'anomaly' if prediction == 1 else 'normal', 'confidence': float(max(proba))}))",
  "useVirtualEnv": true,
  "venvPath": "/home/pi/ml-env",
  "timeout": 30
}

// Input: real-time sensor data
{
  "payload": {
    "temperature": 78.5,
    "vibration": 0.42,
    "pressure": 1013.2
  }
}

// Output: classification result
{
  "class": 1,
  "label": "anomaly",
  "confidence": 0.94
}

Common Use Cases