Script¶

Script node allows users to run custom Python scripts on the device. Due to the computational resource constraints, script node shouldn’t be used for heavy computing (eg. image manipulation/CV), but for managing the flow of the pipeline (business logic). Example use cases would be controlling nodes like ImageManip, ColorCamera, SpatialLocationCalculator, decoding NeuralNetwork results, or interfacing with GPIOs. For debugging scripts, we suggest Script node logging.

How to place it¶

pipeline = dai.Pipeline()
script = pipeline.create(dai.node.Script)

dai::Pipeline pipeline;
auto script = pipeline.create<dai::node::Script>();

Inputs and Outputs¶

inputs[]  ┌──────────────┐  outputs[]
---------►│              ├-----------►
---------►│              ├-----------►
   ...    │    Script    |    ...
   ...    │              │    ...
---------►│              ├-----------►
          └──────────────┘

Users can define as many inputs and outputs as they need. Inputs and outputs can be any Message type.

Usage¶

script = pipeline.create(dai.node.Script)
script.setScript("""
    import time
    import marshal
    num = 123
    node.warn(f"Number {num}") # Print to host
    x = [1, "Hello", {"Foo": "Bar"}]
    x_serial = marshal.dumps(x)
    b = Buffer(len(x_serial))
    while True:
        time.sleep(1)
        b.setData(x_serial)
        node.io['out'].send(b)
""")
script.outputs['out'].link(xout.input)

# ...
# After initializing the device, enable log levels
device.setLogLevel(dai.LogLevel.WARN)
device.setLogOutputLevel(dai.LogLevel.WARN)

auto script = pipeline.create<dai::node::Script>();
script->setScript(R"(
    import time
    import marshal
    num = 123
    node.warn(f"Number {num}") # Print to host
    x = [1, "Hello", {"Foo": "Bar"}]
    x_serial = marshal.dumps(x)
    b = Buffer(len(x_serial))
    while True:
        time.sleep(1)
        b.setData(x_serial)
        node.io['out'].send(b)
)");
script->outputs["out"].link(xout->input);

// ...
// After initializing the device, enable log levels
device.setLogLevel(dai::LogLevel.WARN);
device.setLogOutputLevel(dai::LogLevel.WARN);

Interfacing with GPIOs¶

In the script node you can interface with GPIOs of the VPU using module GPIO. Currently supported functions are:

# Module
import GPIO

# General
GPIO.setup(gpio, dir, pud, exclusive)
GPIO.release(gpio)
GPIO.write(gpio, value)
GPIO.read(gpio)

# Interrupts
GPIO.waitInterruptEvent(gpio = -1) # blocks until any interrupt or interrupt by specified gpio is fired. Interrupts with callbacks are ignored here
GPIO.hasInterruptEvent(gpio = -1) # returns whether interrupt happened on any or specified gpio. Interrupts with callbacks are ignored here
GPIO.setInterrupt(gpio, edge, priority, callback = None) # adds interrupt to specified pin
GPIO.clearInterrupt(gpio) # clears interrupt of specified pin

# PWM
GPIO.setPwm(gpio, highCount, lowCount, repeat=0) # repeat == 0 means indefinite
GPIO.enablePwm(gpio, enable)

# Enumerations
GPIO.Direction: GPIO.IN, GPIO.OUT
GPIO.State: GPIO.LOW, GPIO.HIGH
GPIO.PullDownUp: GPIO.PULL_NONE, GPIO.PULL_DOWN, GPIO.PULL_UP
GPIO.Edge: GPIO.RISING, GPIO.FALLING, GPIO.LEVEL_HIGH, GPIO.LEVEL_LOW

Here’s an example of toggling GPIO pin 40 inside Script node from the host (via XLinkIn). On OAK-SoM-Pro, GPIO 40 drives FSYNC signal for both 4-lane cameras, and we have used the code below for this exact reason.

import GPIO
MX_PIN = 40

ret = GPIO.setup(MX_PIN, GPIO.OUT, GPIO.PULL_DOWN)
toggleVal = True

while True:
  data = node.io['in'].get()  # Wait for a message from the host computer

  node.warn('GPIO toggle: ' + str(toggleVal))
  toggleVal = not toggleVal
  ret = GPIO.write(MX_PIN, toggleVal)  # Toggle the GPIO

Time synchronization¶

Script node has access to both device (internal) clock and also synchronized host clock. Host clock is synchronized with device clock at below 2.5ms precision at 1σ, more information here.

import time
interval = 60
ctrl = CameraControl()
ctrl.setCaptureStill(True)
previous = 0
while True:
    time.sleep(0.001)

    tnow_full = Clock.nowHost() # Synced clock with host
    # Clock.now() -> internal/device clock
    # Clock.offsetToHost() -> Offset between internal/device clock and host clock

    now = tnow_full.seconds
    if now % interval == 0 and now != previous:
        previous = now
        node.warn(f'{tnow_full}')
        node.io['out'].send(ctrl)

Using DepthAI Messages ¶

The depthai module is implicitly imported to the script node. You can create new depthai messages and assign data to it, for example:

buf = Buffer(100) # Assign 100 bytes to the Buffer message

# Create CameraControl message, set manual focus
control = CameraControl()
control.setManualFocus(100)

imgFrame = ImgFrame(300*300*3) # Buffer with 300x300x3 bytes

Available modules and libraries¶

"posix", "errno", "pwd", "_sre", "_codecs", "_weakref", "_functools", "_operator",
"_collections", "_abc", "itertools", "atexit", "_stat", "time", "_datetime", "math",
"_thread", "_io", "_symtable", "marshal", "_ast", "gc", "_warnings", "_string", "_struct"

"binascii", "_random", "_socket", "_md5", "_sha1", "_sha256", "_sha512", "select",
"array", "unicodedata"

"__main__", "_collections_abc", "_frozen_importlib", "_frozen_importlib_external",
"_sitebuiltins", "abc", "codecs", "datetime", "encodings", "encodings.aliases",
"encodings.ascii", "encodings.latin_1", "encodings.mbcs", "encodings.utf_8", "genericpath",
"io", "os", "posixpath", "site", "stat", "threading", "types", "struct", "copyreg",
"reprlib", "operator", "keyword", "heapq", "collections", "functools", "sre_constants",
"sre_parse", "sre_compile", "enum", "re", "json", "json.decoder", "json.encoder",
"json.scanner", "textwrap"

"http", "http.client", "http.server", "html", "mimetypes", "copy", "shutil", "fnmatch",
"socketserver", "contextlib", "email", "email._encoded_words", "email._header_value_parser",
"email._parseaddr", "email._policybase", "email.base64mime", "email.charset",
"email.contentmanager",  "email.encoders", "email.errors", "email.feedparser",
"email.generator", "email.header", "email.headerregistry", "email.iterators", "email.message",
"email.parser", "email.policy", "email.quoprimime", "email.utils", "string", "base64",
"quopri", "random", "warnings", "bisect", "hashlib", "logging", "traceback", "linecache",
"socket", "token", "tokenize", "weakref", "_weakrefset", "collections.abc", "selectors",
"urllib", "urllib.parse", "calendar", "locale", "uu", "encodings.idna", "stringprep"

The difference between module and library is that module is a precompiled C source with Python bindings, whereas library is Python source code packed into a library and precompiled into Python bytecode (before loaded into our Firmware).

Networking/protocol modules/libraries that are available on the LEON_CSS can only be used on OAK POE device. You can specify on which processor the script will run, eg. for LEON_CSS:

script = pipeline.create(dai.node.Script)
script.setProcessor(dai.ProcessorType.LEON_CSS)

Examples of functionality¶

Script camera control - Controlling the camera
Script get local IP - Get local IP
Script HTTP client - Send HTTP request
Script TCP streaming - TCP communication from within Script node, either in host or client mode
Script MQTT publishing - MQTT publishing from within Script node
Script HTTP server - still image over HTTP
Script MJPEG server - MJPEG video stream over HTTP
Script NNData example - Constructs NNData
Triangulation experiment
Movenet decoding (edge mode) - A bit more complex example by geaxgx

Reference¶

class depthai.node.Script

class Id: Node identificator. Unique for every node on a single Pipeline

getAssetManager(*args, **kwargs)

Overloaded function.

getAssetManager(self: depthai.Node) -> depthai.AssetManager
getAssetManager(self: depthai.Node) -> depthai.AssetManager

getInputRefs(*args, **kwargs)

Overloaded function.

getInputRefs(self: depthai.Node) -> list[depthai.Node.Input]
getInputRefs(self: depthai.Node) -> list[depthai.Node.Input]

getInputs(self: depthai.Node) → list[depthai.Node.Input]

getName(self: depthai.Node) → str

getOutputRefs(*args, **kwargs)

Overloaded function.

getOutputRefs(self: depthai.Node) -> list[depthai.Node.Output]
getOutputRefs(self: depthai.Node) -> list[depthai.Node.Output]

getOutputs(self: depthai.Node) → list[depthai.Node.Output]

getParentPipeline(*args, **kwargs)

Overloaded function.

getParentPipeline(self: depthai.Node) -> depthai.Pipeline
getParentPipeline(self: depthai.Node) -> depthai.Pipeline

getProcessor(self: depthai.node.Script) → depthai.ProcessorType

getScriptName(self: depthai.node.Script) → str

setProcessor(self: depthai.node.Script, arg0: depthai.ProcessorType) → None

setScript(*args, **kwargs)

Overloaded function.

setScript(self: depthai.node.Script, script: str, name: str = ‘’) -> None
setScript(self: depthai.node.Script, data: list[int], name: str = ‘’) -> None

setScriptPath(*args, **kwargs)

Overloaded function.

setScriptPath(self: depthai.node.Script, arg0: Path, arg1: str) -> None
setScriptPath(self: depthai.node.Script, path: Path, name: str = ‘’) -> None

class dai::node::Script : public dai::NodeCRTP<Node, Script, ScriptProperties>¶

Public Functions

Script(const std::shared_ptr<PipelineImpl> &par, int64_t nodeId)¶

Script(const std::shared_ptr<PipelineImpl> &par, int64_t nodeId, std::unique_ptr<Properties> props)¶

void setScriptPath(const dai::Path &path, const std::string &name = "")¶

Specify local filesystem path to load the script

Parameters

path: Filesystem path to load the script
name: Optionally set a name of this script, otherwise the name defaults to the path

void setScript(const std::string &script, const std::string &name = "")¶

Sets script data to be interpreted

Parameters

script: Script string to be interpreted
name: Optionally set a name of this script

void setScript(const std::vector<std::uint8_t> &data, const std::string &name = "")¶

Sets script data to be interpreted

Parameters

data: Binary data that represents the script to be interpreted
name: Optionally set a name of this script

dai::Path getScriptPath() const¶

Get filesystem path from where script was loaded.

Return: dai::Path from where script was loaded, otherwise returns empty path

std::string getScriptName() const¶

Get the script name in utf-8.

When name set with setScript() or setScriptPath(), returns that name. When script loaded with setScriptPath() with name not provided, returns the utf-8 string of that path. Otherwise, returns “<script>”

Return: std::string of script name in utf-8

void setProcessor(ProcessorType type)¶

Set on which processor the script should run

Parameters

type: Processor type - Leon CSS or Leon MSS

ProcessorType getProcessor() const¶

Get on which processor the script should run

Return: Processor type - Leon CSS or Leon MSS

Public Members

InputMap inputs¶: Inputs to Script node. Can be accessed using subscript operator (Eg: inputs[‘in1’]) By default inputs are set to blocking with queue size 8

OutputMap outputs¶: Outputs from Script node. Can be accessed subscript operator (Eg: outputs[‘out1’])

Public Static Attributes

static constexpr const char *NAME = "Script"¶

Private Members

dai::Path scriptPath¶

Got questions?

Head over to Discussion Forum for technical support or any other questions you might have.

Script¶

How to place it¶

Inputs and Outputs¶

Usage¶

Interfacing with GPIOs¶

Time synchronization¶

Using DepthAI Messages¶

Available modules and libraries¶

Examples of functionality¶

Reference¶

Got questions?

Using DepthAI Messages ¶