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¶
Also available on GitHub
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | #!/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
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | #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; } |