Script HTTP server

Note

This example can only run on OAK POE devices. You need bootloader on/above version 0.0.15. You can flash bootloader by running python3 examples/bootloader/flash_bootloader.py.

This demo runs an HTTP server on the device itself. Server will serve you a static image when you connect to it (send a GET request).

Demo

When you run the demo, it will print something similar to

Serving at 192.168.1.193:8080

If you open this IP in the browser (eg. chrome), you will see this:

if you click on the here href, you will get a static image. For video stream, you should check out Script MJPEG server.

Setup

Please run the install script to download all required dependencies. Please note that this script must be ran from git context, so you have to download the depthai-python repository first and then run the script

git clone https://github.com/luxonis/depthai-python.git
cd depthai-python/examples
python3 install_requirements.py

For additional information, please follow installation guide

Source code

Python

C++

Also available on GitHub

#!/usr/bin/env python3

import depthai as dai
import time

# Start defining a pipeline
pipeline = dai.Pipeline()

# Define a source - color camera
cam = pipeline.create(dai.node.ColorCamera)
# VideoEncoder
jpeg = pipeline.create(dai.node.VideoEncoder)
jpeg.setDefaultProfilePreset(cam.getFps(), dai.VideoEncoderProperties.Profile.MJPEG)

# Script node
script = pipeline.create(dai.node.Script)
script.setProcessor(dai.ProcessorType.LEON_CSS)
script.setScript("""
    from http.server import BaseHTTPRequestHandler
    import socketserver
    import socket
    import fcntl
    import struct

    PORT = 8080
    ctrl = CameraControl()
    ctrl.setCaptureStill(True)

    def get_ip_address(ifname):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        return socket.inet_ntoa(fcntl.ioctl(
            s.fileno(),
            -1071617759,  # SIOCGIFADDR
            struct.pack('256s', ifname[:15].encode())
        )[20:24])

    class HTTPHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path == '/':
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b'<h1>[DepthAI] Hello, world!</h1><p>Click <a href="img">here</a> for an image</p>')
            elif self.path == '/img':
                node.io['out'].send(ctrl)
                jpegImage = node.io['jpeg'].get()
                self.send_response(200)
                self.send_header('Content-Type', 'image/jpeg')
                self.send_header('Content-Length', str(len(jpegImage.getData())))
                self.end_headers()
                self.wfile.write(jpegImage.getData())
            else:
                self.send_response(404)
                self.end_headers()
                self.wfile.write(b'Url not found...')

    with socketserver.TCPServer(("", PORT), HTTPHandler) as httpd:
        node.warn(f"Serving at {get_ip_address('re0')}:{PORT}")
        httpd.serve_forever()
""")

# Connections
cam.still.link(jpeg.input)
script.outputs['out'].link(cam.inputControl)
jpeg.bitstream.link(script.inputs['jpeg'])

# Connect to device with pipeline
with dai.Device(pipeline) as device:
    while not device.isClosed():
        time.sleep(1)

Also available on GitHub

#include <chrono>
#include <iostream>
#include <thread>

// Includes common necessary includes for development using depthai library
#include "depthai/depthai.hpp"

int main() {
    using namespace std;

    // Start defining a pipeline
    dai::Pipeline pipeline;

    auto cam = pipeline.create<dai::node::ColorCamera>();

    auto jpeg = pipeline.create<dai::node::VideoEncoder>();
    jpeg->setDefaultProfilePreset(cam->getFps(), dai::VideoEncoderProperties::Profile::MJPEG);

    // Script node
    auto script = pipeline.create<dai::node::Script>();
    script->setProcessor(dai::ProcessorType::LEON_CSS);
    script->setScript(R"(
    from http.server import BaseHTTPRequestHandler
    import socketserver
    import socket
    import fcntl
    import struct

    PORT = 8080
    ctrl = CameraControl()
    ctrl.setCaptureStill(True)

    def get_ip_address(ifname):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        return socket.inet_ntoa(fcntl.ioctl(
            s.fileno(),
            -1071617759,  # SIOCGIFADDR
            struct.pack('256s', ifname[:15].encode())
        )[20:24])

    class HTTPHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path == '/':
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b'<h1>[DepthAI] Hello, world!</h1><p>Click <a href="img">here</a> for an image</p>')
            elif self.path == '/img':
                node.io['out'].send(ctrl)
                jpegImage = node.io['jpeg'].get()
                self.send_response(200)
                self.send_header('Content-Type', 'image/jpeg')
                self.send_header('Content-Length', str(len(jpegImage.getData())))
                self.end_headers()
                self.wfile.write(jpegImage.getData())
            else:
                self.send_response(404)
                self.end_headers()
                self.wfile.write(b'Url not found...')

    with socketserver.TCPServer(("", PORT), HTTPHandler) as httpd:
        node.warn(f"Serving at {get_ip_address('re0')}:{PORT}")
        httpd.serve_forever()
    )");

    cam->still.link(jpeg->input);
    script->outputs["out"].link(cam->inputControl);
    jpeg->bitstream.link(script->inputs["jpeg"]);

    // Connect to device with pipeline
    dai::Device device(pipeline);
    while(!device.isClosed()) {
        this_thread::sleep_for(chrono::milliseconds(1000));
    }
    return 0;
}

Got questions?

We’re always happy to help with code or other questions you might have.

Community Discord

Discussion Forum

Email Support