diff --git a/m8-controller-box/README.md b/m8-controller-box/README.md new file mode 100644 index 000000000..eef5d1bca --- /dev/null +++ b/m8-controller-box/README.md @@ -0,0 +1,153 @@ +# M8 Controller Box (PR1) + +The **M8 Controller Box (PR1)** is a USB-connected expansion module designed for integrating CAN devices, GPIO, relays, and peripheral interfaces into OAK-based systems. + +This document serves as the **primary usage reference** for the M8 Controller Box within `oak-examples`. + +> **Note:** The examples in this directory are intended to run on **OAK4 in standalone mode**. The M8 Controller Box connects directly to the OAK4 device, so these examples are not supported in host-driven peripheral mode. + + +## Key Features + +### CAN Interface + +* Supports **CAN 2.0A and 2.0B** +* Baud rates up to **1 Mbps** +* Native **Linux SocketCAN** interface + +> For more information look at [can-example](./can-example/) + +### USB Audio + +* Integrated **buzzer / 3.5mm audio output** +* Available via internal USB connection + +> For more information look at the [simple-example](./simple-example/) + +### USB Expansion + +* **2× USB-A ports** +* USB 2.0 speeds +* Up to **500mA current limit (shared)** + + +### Serial Interface + +* **1× RS232 interface** + +> See [simple-example](./simple-example/) and modify `main.py` with [library example](https://github.com/luxonis/rp2040_u2if/blob/main/examples/ControllerBox/example_controller_box_serial.py). + +### Isolated Strobe Driver + +* Supports **5–24V strobe lights** +* Electrically isolated output + +> For more information look at the [strobe-relay example](./strobe-relay-example/) + +### GPIO + +* **16× GPIO pins** +* 3.3V logic level +* Reverse voltage protection and ESD protection +* Configurable as input or output + +**Electrical limits:** + +* Total combined current must not exceed **50mA** + +> For more information look at the [simple-example](./simple-example/) + +### Power Relays + +* **4× SPDT latching relays** +* Up to **16A current** +* Maximum **400VAC switching voltage** + +> For relay control, see the [strobe-relay example](./strobe-relay-example/). + +### User Interface + +* **3× physical buttons** +* **3× status LEDs** + +> For more example look at the [simple-example](./simple-example/) + +## Pinout + +Device-level pinout is shown below: + +![M8 Controller Box Schematics](media/schematics.png) + + + +## Example Applications + +The repository includes reference applications demonstrating typical usage. + +### [simple-example](./simple-example/) + +* Blinks LED 1 +* Button 2 toggles LED 2 + + +### [depthai-example](./depthai-example/) + +* Based on Luxonis hand pose detection +* Turns on LED 1 when a hand is detected + + +### [can-example](./can-example/) + +* Monitors button 1 +* Sends CAN frame on press +* Uses Linux SocketCAN (`python-can`) + +### [strobe-relay-example](./strobe-relay-example/) + +* Detects barcodes using `pyzbar` +* On barcode detection it switches relay + +## Running The Examples + +All examples in this directory should be run as OAK apps on the OAK4 device. + +1. Install `oakctl` by following the instructions [here](https://docs.luxonis.com/software-v3/oak-apps/oakctl). +2. Change into the example directory you want to run. +3. Connect to the device and start the app: + +```bash +oakctl connect +oakctl app run . +``` + +If you have a locally connected device, `oakctl app run .` may be enough. Use `oakctl connect ` when targeting a device over the network or when multiple devices are available. + +## Notes + +* GPIO and peripheral control is exposed via the **u2if (USB-to-interfaces) protocol** +* Example applications demonstrate recommended interaction patterns +* Library repository: [rp2040_u2if](https://github.com/luxonis/rp2040_u2if) - Note, library must be used within OAK App, to run on the device - running these on Host will result in following error: +``` +File "rp2040_u2if\examples\example_simple.py", line 5, in + + rp2040.open() + + ~~~~~~~~~~~^^ + + File ".venv\Lib\site-packages\luxonis_u2if\rp2040_u2if.py", line 154, in open + + self._hid.open(self._vid, self._pid, self._serial) + + ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + + File "hid.pyx", line 143, in hid.device.open + +OSError: open failed + +Process finished with exit code 1 +``` + + +## Support + +For integration support or early access features, contact support@luxonis.com diff --git a/m8-controller-box/can-example/.oakappignore b/m8-controller-box/can-example/.oakappignore new file mode 100644 index 000000000..671c0ae52 --- /dev/null +++ b/m8-controller-box/can-example/.oakappignore @@ -0,0 +1,33 @@ +# Python virtual environments +venv/ +.venv/ + +# Node.js +# ignore node_modules, it will be reinstalled in the container +node_modules/ + +# Multimedia files +media/ + +# Documentation +README.md + +# VCS +.git/ +.github/ +.gitlab/ + +# The following files are ignored by default +# uncomment a line if you explicitly need it + +# !*.oakapp + +# Python +# !**/.mypy_cache/ +# !**/.ruff_cache/ + +# IDE files +# !**/.idea +# !**/.vscode +# !**/.zed + diff --git a/m8-controller-box/can-example/README.md b/m8-controller-box/can-example/README.md new file mode 100644 index 000000000..0e715855b --- /dev/null +++ b/m8-controller-box/can-example/README.md @@ -0,0 +1,75 @@ +# M8 CAN Transmission Example (Button Triggered) + +This example demonstrates how to send data over the **CAN bus** from the M8 Controller Box by pressing a physical button. + +The application runs inside a container directly on the device and uses `python-can` with the Linux SocketCAN interface. + +> **Note:** This example works only on OAK4 in standalone mode. It is not supported in host-driven peripheral mode because the M8 Controller Box is connected directly to the OAK4 device. + +## Functionality + +This example performs the following actions: + +- Monitors the **button 1**. +- When the button is pressed, a CAN frame is transmitted over the M8 CAN interface (`can0`). + +This provides a minimal, practical reference for sending CAN messages from a containerized application running on the OAK4 device. + +## Usage + +Running this example requires an OAK4 device with the M8 Controller Box attached. + +## Standalone Mode (RVC4 only) + +Install `oakctl` by following the instructions [here](https://docs.luxonis.com/software-v3/oak-apps/oakctl). + +Then run the example from this directory: + +```bash +oakctl connect +oakctl app run . +``` + +This builds the app from the local `oakapp.toml`, deploys it to the OAK4 device, and starts it in standalone mode. + +## Peripheral Mode + +This mode is not supported for this example. The M8 Controller Box must be attached directly to the OAK4 device running the app. + +## CAN Interface Setup (Required) + +Before running the application, the CAN interface must be configured on the target device where CAN messages should be transmitted. + +Run the following commands on the device: + +```bash +ip link set can0 type can bitrate 500000 +ip link set can0 up +``` + +This: + +- Configures the CAN bitrate to **500 kbps** +- Brings the `can0` interface online + +The bitrate must match the rest of the CAN network. + +## Monitoring CAN Traffic + +To verify transmission, you can listen to CAN traffic on a Linux system using: + +```bash +candump can0 +``` + +When the **button 1** is pressed, a CAN frame will appear on the bus. + +## Use Case + +This example serves as a minimal reference implementation for: + +- Sending CAN messages from the M8 Controller Box +- Integrating containerized applications with CAN-based systems +- Trigger-based CAN communication using GPIO inputs + +It can be extended to transmit structured data, control messages, or sensor outputs over CAN. diff --git a/m8-controller-box/can-example/backend-run.sh b/m8-controller-box/can-example/backend-run.sh new file mode 100644 index 000000000..aa18fc9e4 --- /dev/null +++ b/m8-controller-box/can-example/backend-run.sh @@ -0,0 +1,3 @@ +#!/bin/sh +echo "Starting Backend" +exec python3.12 -u /app/main.py diff --git a/m8-controller-box/can-example/main.py b/m8-controller-box/can-example/main.py new file mode 100644 index 000000000..17aeadc28 --- /dev/null +++ b/m8-controller-box/can-example/main.py @@ -0,0 +1,88 @@ +""" +Controller Box CAN Button IRQ Example +------------------------------------ + +Demonstrates how to send a CAN frame using +ControllerBox button events without polling. +Uses GPIO interrupts instead of continuous polling. +""" + +import time +import can +from luxonis_u2if import ControllerBox + + +# ------------------------------------------------------------ +# Connect to ControllerBox device +# ------------------------------------------------------------ + +box = ControllerBox() + + +# ------------------------------------------------------------ +# CAN configuration +# ------------------------------------------------------------ + +CAN_IFACE = "can0" +CAN_ID = 0x123 +CAN_DATA = [0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88] + +bus = can.interface.Bus(channel=CAN_IFACE, interface="socketcan") + + +def send_can_frame(): + """ + Send a predefined CAN frame. + """ + msg = can.Message( + arbitration_id=CAN_ID, + data=CAN_DATA, + is_extended_id=False, + ) + bus.send(msg) + print(f"Sent CAN: {CAN_IFACE} id=0x{CAN_ID:X} data={CAN_DATA}") + + +# ------------------------------------------------------------ +# Button callback +# ------------------------------------------------------------ + +BUTTON_INDEX = 1 # Button 1 +button_pin = ControllerBox.BUTTON_PINS[BUTTON_INDEX - 1] + +def button_cb(btn, state): + """ + Triggered on button press/release. + + Parameters + ---------- + btn : int + Button index (1..3) + state : bool + True = pressed, False = released + """ + if btn == BUTTON_INDEX and state: # pressed + try: + send_can_frame() + except Exception as e: + print(f"[ERROR] Failed to send CAN frame: {e}") + + +# ------------------------------------------------------------ +# Register callback +# ------------------------------------------------------------ + +box.set_btn_callback(button_cb) + + +print("ControllerBox ready") +print("Press Button 1 to send CAN frame") + + +# ------------------------------------------------------------ +# Main loop +# ------------------------------------------------------------ + +# Nothing required here — button events handled via IRQ +while True: + time.sleep(1) \ No newline at end of file diff --git a/m8-controller-box/can-example/oakapp.toml b/m8-controller-box/can-example/oakapp.toml new file mode 100644 index 000000000..0e1a3bc2d --- /dev/null +++ b/m8-controller-box/can-example/oakapp.toml @@ -0,0 +1,33 @@ +identifier = "com.example.m8-controller-box.can-example" +entrypoint = ["bash", "-c", "/usr/bin/runsvdir -P /etc/service"] +app_version = "1.0.2" +assign_frontend_port = true + +prepare_container = [ + { type = "COPY", source = "./requirements.txt", target = "./requirements.txt" }, + { type = "RUN", command = "python3.12 -m pip install -r /app/requirements.txt --break-system-packages" }, + { type = "RUN", command = "bash -lc 'set -e; apt-get update && apt-get install -y libhidapi-hidraw0 libhidapi-libusb0'" }, +] + +build_steps = [ + "mkdir -p /etc/service/backend", + "cp /app/backend-run.sh /etc/service/backend/run", + "chmod +x /etc/service/backend/run", +] + +allowed_devices = [{ allow = true, access = "rwm" }] + +additional_mounts = [ + { source = "/dev", target = "/dev", type = "devtmpfs", options = [ + "mode=777", + ] }, +] + +[base_image] +api_url = "https://registry-1.docker.io" +service = "registry.docker.io" +oauth_url = "https://auth.docker.io/token" +auth_type = "repository" +auth_name = "luxonis/oakapp-base" +image_name = "luxonis/oakapp-base" +image_tag = "1.2.6" diff --git a/m8-controller-box/can-example/requirements.txt b/m8-controller-box/can-example/requirements.txt new file mode 100644 index 000000000..c8ca5e696 --- /dev/null +++ b/m8-controller-box/can-example/requirements.txt @@ -0,0 +1,3 @@ +hidapi +python-can +git+https://github.com/luxonis/rp2040_u2if.git diff --git a/m8-controller-box/depthai-example/.oakappignore b/m8-controller-box/depthai-example/.oakappignore new file mode 100644 index 000000000..671c0ae52 --- /dev/null +++ b/m8-controller-box/depthai-example/.oakappignore @@ -0,0 +1,33 @@ +# Python virtual environments +venv/ +.venv/ + +# Node.js +# ignore node_modules, it will be reinstalled in the container +node_modules/ + +# Multimedia files +media/ + +# Documentation +README.md + +# VCS +.git/ +.github/ +.gitlab/ + +# The following files are ignored by default +# uncomment a line if you explicitly need it + +# !*.oakapp + +# Python +# !**/.mypy_cache/ +# !**/.ruff_cache/ + +# IDE files +# !**/.idea +# !**/.vscode +# !**/.zed + diff --git a/m8-controller-box/depthai-example/README.md b/m8-controller-box/depthai-example/README.md new file mode 100644 index 000000000..3159964ee --- /dev/null +++ b/m8-controller-box/depthai-example/README.md @@ -0,0 +1,40 @@ +# Hand Pose Example with M8 Controller LED Trigger + +This project is based on the official Luxonis hand pose example: + +https://github.com/luxonis/oak-examples/tree/main/neural-networks/pose-estimation/hand-pose + +It demonstrates hand detection and hand landmark estimation using DepthAI. + +> **Note:** This example works only on OAK4 in standalone mode. It is not supported in host-driven peripheral mode because the M8 Controller Box is connected directly to the OAK4 device. + +## Added Functionality + +In addition to the original example, this version integrates an **M8 Controller Box**. + +When a hand is detected: + +- **LED 1** on the M8 controller will turn on. + +This allows simple hardware feedback triggered directly from hand detection events. + +## Usage + +Running this example requires an OAK4 device with the M8 Controller Box attached. + +## Standalone Mode (RVC4 only) + +Install `oakctl` by following the instructions [here](https://docs.luxonis.com/software-v3/oak-apps/oakctl). + +Then run the example from this directory: + +```bash +oakctl connect +oakctl app run . +``` + +This builds the app from the local `oakapp.toml`, deploys it to the OAK4 device, and starts it in standalone mode. + +## Peripheral Mode + +This mode is not supported for this example. The M8 Controller Box must be attached directly to the OAK4 device running the app. diff --git a/m8-controller-box/depthai-example/backend-run.sh b/m8-controller-box/depthai-example/backend-run.sh new file mode 100644 index 000000000..aa18fc9e4 --- /dev/null +++ b/m8-controller-box/depthai-example/backend-run.sh @@ -0,0 +1,3 @@ +#!/bin/sh +echo "Starting Backend" +exec python3.12 -u /app/main.py diff --git a/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC2.yaml b/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC2.yaml new file mode 100644 index 000000000..e8e8089ec --- /dev/null +++ b/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC2.yaml @@ -0,0 +1,2 @@ +model: luxonis/mediapipe-hand-landmarker:224x224 +platform: RVC2 \ No newline at end of file diff --git a/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC4.yaml b/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC4.yaml new file mode 100644 index 000000000..1e61a842f --- /dev/null +++ b/m8-controller-box/depthai-example/depthai_models/mediapipe_hand_landmarker.RVC4.yaml @@ -0,0 +1,2 @@ +model: luxonis/mediapipe-hand-landmarker:224x224 +platform: RVC4 \ No newline at end of file diff --git a/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC2.yaml b/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC2.yaml new file mode 100644 index 000000000..cc87d00a2 --- /dev/null +++ b/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC2.yaml @@ -0,0 +1,2 @@ +model: luxonis/mediapipe-palm-detection:192x192 +platform: RVC2 \ No newline at end of file diff --git a/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC4.yaml b/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC4.yaml new file mode 100644 index 000000000..ee0ab6c0f --- /dev/null +++ b/m8-controller-box/depthai-example/depthai_models/mediapipe_palm_detection.RVC4.yaml @@ -0,0 +1,2 @@ +model: luxonis/mediapipe-palm-detection:192x192 +platform: RVC4 \ No newline at end of file diff --git a/m8-controller-box/depthai-example/main.py b/m8-controller-box/depthai-example/main.py new file mode 100644 index 000000000..ef5fd42bb --- /dev/null +++ b/m8-controller-box/depthai-example/main.py @@ -0,0 +1,211 @@ +""" +Controller Box DepthAI Hand Detection Example +--------------------------------------------- + +This example demonstrates integration of the ControllerBox +with a DepthAI hand detection pipeline. + +- Runs hand detection using DepthAI +- Turns ON LED 1 when a hand is detected +- Turns OFF LED 1 when no hand is present + +This showcases how to combine AI perception with hardware control. +""" + +from pathlib import Path +import time + +import depthai as dai +from depthai_nodes.node import ParsingNeuralNetwork, GatherData + +from utils.arguments import initialize_argparser +from utils.annotation_node import AnnotationNode +from utils.process import ProcessDetections + +# Controller Box +from luxonis_u2if import ControllerBox + + +# ------------------------------------------------------------ +# Connect to ControllerBox device +# ------------------------------------------------------------ + +box = ControllerBox() +box.led_init() + + +# ------------------------------------------------------------ +# Configuration +# ------------------------------------------------------------ + +_, args = initialize_argparser() + +PADDING = 0.1 +CONFIDENCE_THRESHOLD = 0.5 + + +# ------------------------------------------------------------ +# Device setup +# ------------------------------------------------------------ + +visualizer = dai.RemoteConnection(httpPort=8082) +device = dai.Device(dai.DeviceInfo(args.device)) if args.device else dai.Device() +platform = device.getPlatform().name + +print(f"Platform: {platform}") + +frame_type = ( + dai.ImgFrame.Type.BGR888p if platform == "RVC2" else dai.ImgFrame.Type.BGR888i +) + +if not args.fps_limit: + args.fps_limit = 8 if platform == "RVC2" else 30 + print( + f"\nFPS limit set to {args.fps_limit} for {platform} platform.\n" + ) + + +# ------------------------------------------------------------ +# Pipeline +# ------------------------------------------------------------ + +with dai.Pipeline(device) as pipeline: + print("Creating pipeline...") + + # Detection model + det_model_description = dai.NNModelDescription.fromYamlFile( + f"mediapipe_palm_detection.{platform}.yaml" + ) + det_nn_archive = dai.NNArchive(dai.getModelFromZoo(det_model_description)) + + # Pose model + pose_model_description = dai.NNModelDescription.fromYamlFile( + f"mediapipe_hand_landmarker.{platform}.yaml" + ) + pose_nn_archive = dai.NNArchive(dai.getModelFromZoo(pose_model_description)) + + # Input source + if args.media_path: + replay = pipeline.create(dai.node.ReplayVideo) + replay.setReplayVideoFile(Path(args.media_path)) + replay.setOutFrameType(frame_type) + replay.setLoop(True) + if args.fps_limit: + replay.setFps(args.fps_limit) + input_node = replay.out + else: + cam = pipeline.create(dai.node.Camera).build() + cam_out = cam.requestOutput((768, 768), frame_type, fps=args.fps_limit) + input_node = cam_out + + # Resize for detection + resize_node = pipeline.create(dai.node.ImageManip) + resize_node.setMaxOutputFrameSize( + det_nn_archive.getInputWidth() * det_nn_archive.getInputHeight() * 3 + ) + resize_node.initialConfig.setOutputSize( + det_nn_archive.getInputWidth(), + det_nn_archive.getInputHeight(), + mode=dai.ImageManipConfig.ResizeMode.STRETCH, + ) + resize_node.initialConfig.setFrameType(frame_type) + input_node.link(resize_node.inputImage) + + detection_nn: ParsingNeuralNetwork = pipeline.create(ParsingNeuralNetwork).build( + resize_node.out, det_nn_archive + ) + + # Detection processing + detections_processor = pipeline.create(ProcessDetections).build( + detections_input=detection_nn.out, + padding=PADDING, + target_size=(pose_nn_archive.getInputWidth(), pose_nn_archive.getInputHeight()), + ) + + # Script node + script = pipeline.create(dai.node.Script) + script.setScriptPath(str(Path(__file__).parent / "utils/script.py")) + + detection_nn.passthrough.link(script.inputs["frame_input"]) + detections_processor.config_output.link(script.inputs["config_input"]) + detections_processor.num_configs_output.link(script.inputs["num_configs_input"]) + + # Pose manipulation + pose_manip = pipeline.create(dai.node.ImageManip) + pose_manip.initialConfig.setOutputSize( + pose_nn_archive.getInputWidth(), + pose_nn_archive.getInputHeight(), + ) + + script.outputs["output_config"].link(pose_manip.inputConfig) + script.outputs["output_frame"].link(pose_manip.inputImage) + + pose_nn: ParsingNeuralNetwork = pipeline.create(ParsingNeuralNetwork).build( + pose_manip.out, pose_nn_archive + ) + + # Sync detections + pose + gather_data = pipeline.create(GatherData).build(camera_fps=args.fps_limit) + detection_nn.out.link(gather_data.input_reference) + pose_nn.outputs.link(gather_data.input_data) + + # Annotation + connection_pairs = ( + pose_nn_archive.getConfig() + .model.heads[0] + .metadata.extraParams["skeleton_edges"] + ) + + annotation_node = pipeline.create(AnnotationNode).build( + gathered_data=gather_data.out, + video=input_node, + padding_factor=PADDING, + confidence_threshold=CONFIDENCE_THRESHOLD, + connections_pairs=connection_pairs, + ) + + # Video encoding + video_encode_manip = pipeline.create(dai.node.ImageManip) + video_encode_manip.initialConfig.setOutputSize(768, 768) + video_encode_manip.initialConfig.setFrameType(dai.ImgFrame.Type.NV12) + input_node.link(video_encode_manip.inputImage) + + video_encoder = pipeline.create(dai.node.VideoEncoder) + video_encoder.setDefaultProfilePreset( + args.fps_limit, dai.VideoEncoderProperties.Profile.H264_MAIN + ) + video_encode_manip.out.link(video_encoder.input) + + # Visualization + visualizer.addTopic("Video", video_encoder.out, "images") + visualizer.addTopic("Detections", annotation_node.out_detections, "images") + visualizer.addTopic("Pose", annotation_node.out_pose_annotations, "images") + + print("Pipeline created.") + + # Output queue + det_queue = detection_nn.out.createOutputQueue(maxSize=4, blocking=False) + + pipeline.start() + visualizer.registerPipeline(pipeline) + + + # ------------------------------------------------------------ + # Main loop + # ------------------------------------------------------------ + + while pipeline.isRunning(): + det_in = det_queue.tryGet() + + if det_in is not None: + detections = det_in.detections if hasattr(det_in, "detections") else [] + + if len(detections) > 0: + box.led_on(1) + else: + box.led_off(1) + + key = visualizer.waitKey(1) + if key == ord("q"): + print("Got q key. Exiting...") + break \ No newline at end of file diff --git a/m8-controller-box/depthai-example/oakapp.toml b/m8-controller-box/depthai-example/oakapp.toml new file mode 100644 index 000000000..325a0ff74 --- /dev/null +++ b/m8-controller-box/depthai-example/oakapp.toml @@ -0,0 +1,35 @@ +identifier = "com.example.m8-controller-box.depthai-example" +entrypoint = ["bash", "-c", "/usr/bin/runsvdir -P /etc/service"] +app_version = "1.0.2" +assign_frontend_port = true + +prepare_container = [ + { type = "COPY", source = "./requirements.txt", target = "./requirements.txt" }, + { type = "RUN", command = "python3.12 -m pip install -r /app/requirements.txt --break-system-packages"}, + { type = "RUN", command = "bash -lc 'set -e; apt-get update && apt-get install -y libhidapi-hidraw0 libhidapi-libusb0'" } +] + +build_steps = [ + "mkdir -p /etc/service/backend", + "cp /app/backend-run.sh /etc/service/backend/run", + "chmod +x /etc/service/backend/run", +] + +depthai_models = { yaml_path = "./depthai_models" } + +allowed_devices = [{ allow = true, access = "rwm" }] + +additional_mounts = [ + { source = "/dev", target = "/dev", type = "devtmpfs", options = [ + "mode=777", + ] } +] + +[base_image] +api_url = "https://registry-1.docker.io" +service = "registry.docker.io" +oauth_url = "https://auth.docker.io/token" +auth_type = "repository" +auth_name = "luxonis/oakapp-base" +image_name = "luxonis/oakapp-base" +image_tag = "1.2.6" diff --git a/m8-controller-box/depthai-example/requirements.txt b/m8-controller-box/depthai-example/requirements.txt new file mode 100644 index 000000000..c8078b323 --- /dev/null +++ b/m8-controller-box/depthai-example/requirements.txt @@ -0,0 +1,4 @@ +depthai==3.0.0 +depthai-nodes==0.3.4 +hidapi +git+https://github.com/luxonis/rp2040_u2if.git diff --git a/m8-controller-box/depthai-example/utils/__init__.py b/m8-controller-box/depthai-example/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/m8-controller-box/depthai-example/utils/annotation_node.py b/m8-controller-box/depthai-example/utils/annotation_node.py new file mode 100644 index 000000000..18efc36f1 --- /dev/null +++ b/m8-controller-box/depthai-example/utils/annotation_node.py @@ -0,0 +1,137 @@ +import depthai as dai +from depthai_nodes import ( + ImgDetectionsExtended, + ImgDetectionExtended, + Keypoints, + Predictions, + GatheredData, + SECONDARY_COLOR, +) +from depthai_nodes.utils import AnnotationHelper +from typing import List +from utils.gesture_recognition import recognize_gesture + + +class AnnotationNode(dai.node.HostNode): + def __init__(self) -> None: + super().__init__() + self.gathered_data = self.createInput() + self.out_detections = self.createOutput() + self.out_pose_annotations = self.createOutput( + possibleDatatypes=[ + dai.Node.DatatypeHierarchy(dai.DatatypeEnum.ImgAnnotations, True) + ] + ) + self.confidence_threshold = 0.5 + self.padding_factor = 0.1 + self.connection_pairs = [[]] + + def build( + self, + gathered_data: dai.Node.Output, + video: dai.Node.Output, + confidence_threshold: float, + padding_factor: float, + connections_pairs: List[List[int]], + ) -> "AnnotationNode": + self.confidence_threshold = confidence_threshold + self.padding_factor = padding_factor + self.connection_pairs = connections_pairs + self.link_args(gathered_data, video) + return self + + def process(self, gathered_data: dai.Buffer, video_message: dai.ImgFrame) -> None: + assert isinstance(gathered_data, GatheredData) + + detections_message: ImgDetectionsExtended = gathered_data.reference_data + detections_list: List[ImgDetectionExtended] = detections_message.detections + + new_dets = ImgDetectionsExtended() + new_dets.transformation = video_message.getTransformation() + + annotation_helper = AnnotationHelper() + + for ix, detection in enumerate(detections_list): + keypoints_msg: Keypoints = gathered_data.gathered[ix]["0"] + confidence_msg: Predictions = gathered_data.gathered[ix]["1"] + handness_msg: Predictions = gathered_data.gathered[ix]["2"] + + hand_confidence = confidence_msg.prediction + handness = handness_msg.prediction + + if hand_confidence < self.confidence_threshold: + continue + + width = detection.rotated_rect.size.width + height = detection.rotated_rect.size.height + + xmin = detection.rotated_rect.center.x - width / 2 + xmax = detection.rotated_rect.center.x + width / 2 + ymin = detection.rotated_rect.center.y - height / 2 + ymax = detection.rotated_rect.center.y + height / 2 + + padding = self.padding_factor + + slope_x = (xmax + padding) - (xmin - padding) + slope_y = (ymax + padding) - (ymin - padding) + + new_det = ImgDetectionExtended() + new_det.rotated_rect = ( + detection.rotated_rect.center.x, + detection.rotated_rect.center.y, + detection.rotated_rect.size.width + 2 * padding, + detection.rotated_rect.size.height + 2 * padding, + detection.rotated_rect.angle, + ) + new_det.label = 0 + new_det.label_name = "Hand" + new_det.confidence = detection.confidence + new_dets.detections.append(new_det) + + xs = [] + ys = [] + + for kp in keypoints_msg.keypoints: + x = min(max(xmin - padding + slope_x * kp.x, 0.0), 1.0) + y = min(max(ymin - padding + slope_y * kp.y, 0.0), 1.0) + xs.append(x) + ys.append(y) + + for connection in self.connection_pairs: + pt1_ix, pt2_ix = connection + annotation_helper.draw_line( + pt1=(xs[pt1_ix], ys[pt1_ix]), + pt2=(xs[pt2_ix], ys[pt2_ix]), + ) + + keypoints = [[kpt[0], kpt[1]] for kpt in zip(xs, ys)] + + gesture = recognize_gesture(keypoints) + + text = "Left" if handness < 0.5 else "Right" + text += f" {gesture}" + + text_x = detection.rotated_rect.center.x - 0.05 + text_y = detection.rotated_rect.center.y - height / 2 - 0.10 + + annotation_helper.draw_text( + text=text, + position=(text_x, text_y), + color=SECONDARY_COLOR, + size=32, + ) + + annotation_helper.draw_points( + points=keypoints, color=SECONDARY_COLOR, thickness=2 + ) + + new_dets.setTimestamp(detections_message.getTimestamp()) + new_dets.setSequenceNum(detections_message.getSequenceNum()) + self.out_detections.send(new_dets) + + annotations = annotation_helper.build( + timestamp=detections_message.getTimestamp(), + sequence_num=detections_message.getSequenceNum(), + ) + + self.out_pose_annotations.send(annotations) diff --git a/m8-controller-box/depthai-example/utils/arguments.py b/m8-controller-box/depthai-example/utils/arguments.py new file mode 100644 index 000000000..4ef0b9862 --- /dev/null +++ b/m8-controller-box/depthai-example/utils/arguments.py @@ -0,0 +1,38 @@ +import argparse + + +def initialize_argparser(): + """Initialize the argument parser for the script.""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + + parser.add_argument( + "-d", + "--device", + help="Optional name, DeviceID or IP of the camera to connect to.", + required=False, + default=None, + type=str, + ) + + parser.add_argument( + "-fps", + "--fps_limit", + help="FPS limit for the model runtime.", + required=False, + default=None, + type=int, + ) + + parser.add_argument( + "-media", + "--media_path", + help="Path to the media file you aim to run the model on. If not set, the model will run on the camera input.", + required=False, + default=None, + type=str, + ) + args = parser.parse_args() + + return parser, args diff --git a/m8-controller-box/depthai-example/utils/gesture_recognition.py b/m8-controller-box/depthai-example/utils/gesture_recognition.py new file mode 100644 index 000000000..381e369f5 --- /dev/null +++ b/m8-controller-box/depthai-example/utils/gesture_recognition.py @@ -0,0 +1,132 @@ +import numpy as np +from typing import List, Tuple + + +def distance(a, b): + return np.linalg.norm(a - b) + + +def angle(a, b, c): + ba = a - b + bc = c - b + cosine_angle = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc)) + angle = np.arccos(cosine_angle) + + return np.degrees(angle) + + +def recognize_gesture(kpts: List[Tuple[float, float]]) -> str: + kpts = np.array(kpts) + d_3_5 = distance(kpts[3], kpts[5]) + d_2_3 = distance(kpts[2], kpts[3]) + angle0 = angle(kpts[0], kpts[1], kpts[2]) + angle1 = angle(kpts[1], kpts[2], kpts[3]) + angle2 = angle(kpts[2], kpts[3], kpts[4]) + thumb_state = 0 + index_state = 0 + middle_state = 0 + ring_state = 0 + little_state = 0 + gesture = None + if angle0 + angle1 + angle2 > 460 and d_3_5 / d_2_3 > 1.2: + thumb_state = 1 + else: + thumb_state = 0 + + if kpts[8][1] < kpts[7][1] < kpts[6][1]: + index_state = 1 + elif kpts[6][1] < kpts[8][1]: + index_state = 0 + else: + index_state = -1 + + if kpts[12][1] < kpts[11][1] < kpts[10][1]: + middle_state = 1 + elif kpts[10][1] < kpts[12][1]: + middle_state = 0 + else: + middle_state = -1 + + if kpts[16][1] < kpts[15][1] < kpts[14][1]: + ring_state = 1 + elif kpts[14][1] < kpts[16][1]: + ring_state = 0 + else: + ring_state = -1 + + if kpts[20][1] < kpts[19][1] < kpts[18][1]: + little_state = 1 + elif kpts[18][1] < kpts[20][1]: + little_state = 0 + else: + little_state = -1 + + # Gesture + if ( + thumb_state == 1 + and index_state == 1 + and middle_state == 1 + and ring_state == 1 + and little_state == 1 + ): + gesture = "FIVE" + elif ( + thumb_state == 0 + and index_state == 0 + and middle_state == 0 + and ring_state == 0 + and little_state == 0 + ): + gesture = "FIST" + elif ( + thumb_state == 1 + and index_state == 0 + and middle_state == 0 + and ring_state == 0 + and little_state == 0 + ): + gesture = "OK" + elif ( + thumb_state == 0 + and index_state == 1 + and middle_state == 1 + and ring_state == 0 + and little_state == 0 + ): + gesture = "PEACE" + elif ( + thumb_state == 0 + and index_state == 1 + and middle_state == 0 + and ring_state == 0 + and little_state == 0 + ): + gesture = "ONE" + elif ( + thumb_state == 1 + and index_state == 1 + and middle_state == 0 + and ring_state == 0 + and little_state == 0 + ): + gesture = "TWO" + elif ( + thumb_state == 1 + and index_state == 1 + and middle_state == 1 + and ring_state == 0 + and little_state == 0 + ): + gesture = "THREE" + elif ( + thumb_state == 0 + and index_state == 1 + and middle_state == 1 + and ring_state == 1 + and little_state == 1 + ): + gesture = "FOUR" + else: + gesture = None + + return gesture diff --git a/m8-controller-box/depthai-example/utils/process.py b/m8-controller-box/depthai-example/utils/process.py new file mode 100644 index 000000000..9db495bf2 --- /dev/null +++ b/m8-controller-box/depthai-example/utils/process.py @@ -0,0 +1,78 @@ +import depthai as dai +from depthai_nodes import ImgDetectionsExtended, ImgDetectionExtended +from typing import Tuple + + +class ProcessDetections(dai.node.HostNode): + """A host node for processing a list of detections in a two-stage pipeline. + The node iterates over a list of detections and sends a dai.MessageGroup with + a list of ImageManipConfig objects that can be executed by the ImageManip node. + + Before use, the target size need to be set with the set_target_size method. + Attributes + ---------- + detections_input : dai.Input + The input message for the detections. + config_output : dai.Output + The output message for the ImageManipConfig objects. + num_configs_output : dai.Output + The output message for the number of configs. + padding: float + The padding factor to enlarge the bounding box a little bit. + + """ + + def __init__(self): + super().__init__() + self.detections_input = self.createInput() + self.config_output = self.createOutput() + self.num_configs_output = self.createOutput() + self.padding = 0.1 + self._target_h = None + self._target_w = None + + def build( + self, + detections_input: dai.Node.Output, + padding: float, + target_size: Tuple[int, int], + ) -> "ProcessDetections": + self.padding = padding + self._target_w = target_size[0] + self._target_h = target_size[1] + self.link_args(detections_input) + return self + + def process(self, img_detections: dai.Buffer) -> None: + assert isinstance(img_detections, ImgDetectionsExtended) + detections = img_detections.detections + + num_detections = len(detections) + num_cfgs_message = dai.Buffer(num_detections) + + num_cfgs_message.setTimestamp(img_detections.getTimestamp()) + num_cfgs_message.setSequenceNum(img_detections.getSequenceNum()) + self.num_configs_output.send(num_cfgs_message) + + for i, detection in enumerate(detections): + cfg = dai.ImageManipConfig() + detection: ImgDetectionExtended = detection + rect = detection.rotated_rect + + new_rect = dai.RotatedRect() + new_rect.center.x = rect.center.x + new_rect.center.y = rect.center.y + new_rect.size.width = rect.size.width + 0.1 * 2 + new_rect.size.height = rect.size.height + 0.1 * 2 + new_rect.angle = 0 + + cfg.addCropRotatedRect(new_rect, normalizedCoords=True) + cfg.setOutputSize( + self._target_w, + self._target_h, + dai.ImageManipConfig.ResizeMode.STRETCH, + ) + cfg.setReusePreviousImage(False) + cfg.setTimestamp(img_detections.getTimestamp()) + cfg.setSequenceNum(img_detections.getSequenceNum()) + self.config_output.send(cfg) diff --git a/m8-controller-box/depthai-example/utils/script.py b/m8-controller-box/depthai-example/utils/script.py new file mode 100644 index 000000000..5746ce683 --- /dev/null +++ b/m8-controller-box/depthai-example/utils/script.py @@ -0,0 +1,23 @@ +try: + while True: + frame = node.inputs["frame_input"].get() + # node.warn(f"{frame.getType()}") + # node.warn(f"[ConfigSender {frame.getSequenceNum()}] Got frame {frame.getTimestamp()}") + num_configs_message = node.inputs["num_configs_input"].get() + conf_seq = num_configs_message.getSequenceNum() + frame_seq = frame.getSequenceNum() + num_configs = len(bytearray(num_configs_message.getData())) + + while conf_seq > frame_seq: + # node.warn(f"[ConfigSender {conf_seq}] Configs {conf_seq} mismatch with frame {frame_seq}") + frame = node.inputs["frame_input"].get() + + for i in range(num_configs): + cfg = node.inputs["config_input"].get() + # node.warn(f"[ConfigSender {conf_seq}] Got config {i}") + node.outputs["output_config"].send(cfg) + node.outputs["output_frame"].send(frame) + # node.warn(f"[ConfigSender {conf_seq}] sent {i}") + +except Exception as e: + node.warn(str(e)) diff --git a/m8-controller-box/media/schematics.png b/m8-controller-box/media/schematics.png new file mode 100644 index 000000000..27791a8c8 Binary files /dev/null and b/m8-controller-box/media/schematics.png differ diff --git a/m8-controller-box/simple-example/.oakappignore b/m8-controller-box/simple-example/.oakappignore new file mode 100644 index 000000000..671c0ae52 --- /dev/null +++ b/m8-controller-box/simple-example/.oakappignore @@ -0,0 +1,33 @@ +# Python virtual environments +venv/ +.venv/ + +# Node.js +# ignore node_modules, it will be reinstalled in the container +node_modules/ + +# Multimedia files +media/ + +# Documentation +README.md + +# VCS +.git/ +.github/ +.gitlab/ + +# The following files are ignored by default +# uncomment a line if you explicitly need it + +# !*.oakapp + +# Python +# !**/.mypy_cache/ +# !**/.ruff_cache/ + +# IDE files +# !**/.idea +# !**/.vscode +# !**/.zed + diff --git a/m8-controller-box/simple-example/README.md b/m8-controller-box/simple-example/README.md new file mode 100644 index 000000000..54b4d19ab --- /dev/null +++ b/m8-controller-box/simple-example/README.md @@ -0,0 +1,36 @@ +# Minimal M8 Controller Box Container Example + +This is a minimal example showing how to run an OAK app that controls the **M8 Controller Box** directly from an **OAK4** device. + +> **Note:** This example works only on OAK4 in standalone mode. It is not supported in host-driven peripheral mode because the M8 Controller Box is connected directly to the OAK4 device. + +## Functionality + +This example performs simple GPIO interactions: + +- The LED connected to **pin 18** blinks continuously. +- When the button connected to **pin 19** is pressed: + - The LED connected to **pin 17** turns on. + +This provides a minimal reference setup for applications running directly on the OAK4 device with the M8 Controller Box attached. + +## Usage + +Running this example requires an OAK4 device with the M8 Controller Box attached. + +## Standalone Mode (RVC4 only) + +Install `oakctl` by following the instructions [here](https://docs.luxonis.com/software-v3/oak-apps/oakctl). + +Then run the example from this directory: + +```bash +oakctl connect +oakctl app run . +``` + +This builds the app from the local `oakapp.toml`, deploys it to the OAK4 device, and starts it in standalone mode. + +## Peripheral Mode + +This mode is not supported for this example. The M8 Controller Box must be attached directly to the OAK4 device running the app. diff --git a/m8-controller-box/simple-example/backend-run.sh b/m8-controller-box/simple-example/backend-run.sh new file mode 100644 index 000000000..aa18fc9e4 --- /dev/null +++ b/m8-controller-box/simple-example/backend-run.sh @@ -0,0 +1,3 @@ +#!/bin/sh +echo "Starting Backend" +exec python3.12 -u /app/main.py diff --git a/m8-controller-box/simple-example/main.py b/m8-controller-box/simple-example/main.py new file mode 100644 index 000000000..837e12d0b --- /dev/null +++ b/m8-controller-box/simple-example/main.py @@ -0,0 +1,123 @@ +""" +Controller Box Simple Example – Button Beep & LED +------------------------------------------------- + +This simple example demonstrates core functionality of the M8 Controller Box with a USB audio hub: + +1. **LED Blinking**: LED 1 blinks continuously to indicate the program is running. +2. **Button Handling via GPIO IRQ**: Button connected to GPIO pin 20 triggers events without polling. +3. **Continuous Beep on Button Hold**: While the button is pressed, a 1 kHz tone plays continuously through the USB audio speaker. +4. **LED During Button Press**: LED 2 (or same LED if desired) lights only while the button is pressed. +5. **Minimal Dependencies**: Works in a minimal Linux container using only Python and `aplay`. No additional packages required. + +This example is ideal for understanding basic GPIO interaction, interrupt-driven button handling, and simple audio output. +""" + +import time +import os +from luxonis_u2if import ControllerBox + +# ------------------------------------------------------------ +# Connect to the ControllerBox +# ------------------------------------------------------------ +box = ControllerBox() +box.led_init() # Initialize all LEDs + +# ------------------------------------------------------------ +# Button setup (IRQ) +# ------------------------------------------------------------ +BUTTON_PIN = 20 # GPIO pin where the button is connected +box.gpio_init(BUTTON_PIN, box.GPIO_IN, box.GPIO_PULL_UP) + +# Enable interrupts for rising and falling edges with debounce +box.gpio_set_irq( + BUTTON_PIN, + box.IRQ_RISING | box.IRQ_FALLING, + debounce=True +) + +# ------------------------------------------------------------ +# Audio setup +# ------------------------------------------------------------ +CARD = 1 # USB audio card number (from /proc/asound/cards) +DEVICE = 0 # Audio device on that card +FREQ = 1000 # Beep frequency in Hz +DURATION = 0.1 # Tone chunk duration in seconds +TONE_FILE = "/tmp/beep.wav" # Temporary WAV file to use with aplay + +def generate_wav(): + """Generate a short 1kHz sine wave WAV file if it doesn't exist.""" + import wave, struct, math + framerate = 44100 + amplitude = 32767 + n_samples = int(framerate * DURATION) + + with wave.open(TONE_FILE, 'w') as wf: + wf.setnchannels(1) # Mono + wf.setsampwidth(2) # 16-bit + wf.setframerate(framerate) + for i in range(n_samples): + value = int(amplitude * math.sin(2 * math.pi * FREQ * i / framerate)) + wf.writeframesraw(struct.pack('/dev/null 2>&1") + +# ------------------------------------------------------------ +# Program ready message +# ------------------------------------------------------------ +print("ControllerBox Simple Example ready.") +print("LED 1 blinks continuously.") +print("Press and hold the button to light the LED and play a beep.\n") + +# ------------------------------------------------------------ +# Main loop +# ------------------------------------------------------------ +blink_interval = 0.5 # Time for LED 1 blink +last_blink = time.monotonic() +led_on = False +button_pressed = False # Track if button is currently pressed + +while True: + now = time.monotonic() + + # ---------------------------- + # LED 1 blinking logic + # ---------------------------- + if now - last_blink >= blink_interval: + led_on = not led_on + if led_on: + box.led_on(1) + else: + box.led_off(1) + last_blink = now + + # ---------------------------- + # Handle button IRQ events + # ---------------------------- + for pin, event in box.gpio_get_irq(): + if pin != BUTTON_PIN: + continue + + if event == box.IRQ_RISING: + # Button pressed + button_pressed = True + box.led_on(2) # Turn on LED 2 while pressed + print("Button pressed → starting beep") + + elif event == box.IRQ_FALLING: + # Button released + button_pressed = False + box.led_off(2) # Turn off LED 2 + print("Button released → stopping beep") + + # ---------------------------- + # Continuous beep while button held + # ---------------------------- + if button_pressed: + play_tone() # Play short tone repeatedly + + time.sleep(0.01) # Small delay for loop efficiency diff --git a/m8-controller-box/simple-example/oakapp.toml b/m8-controller-box/simple-example/oakapp.toml new file mode 100644 index 000000000..f33a7025a --- /dev/null +++ b/m8-controller-box/simple-example/oakapp.toml @@ -0,0 +1,33 @@ +identifier = "com.example.m8-controller-box.simple-example" +entrypoint = ["bash", "-c", "/usr/bin/runsvdir -P /etc/service"] +app_version = "1.0.2" +assign_frontend_port = true + +prepare_container = [ + { type = "COPY", source = "./requirements.txt", target = "./requirements.txt" }, + { type = "RUN", command = "python3.12 -m pip install -r /app/requirements.txt --break-system-packages" }, + { type = "RUN", command = "bash -lc 'set -e; apt-get update && apt-get install -y libhidapi-hidraw0 libhidapi-libusb0 alsa-utils'" } +] + +build_steps = [ + "mkdir -p /etc/service/backend", + "cp /app/backend-run.sh /etc/service/backend/run", + "chmod +x /etc/service/backend/run", +] + +allowed_devices = [{ allow = true, access = "rwm" }] + +additional_mounts = [ + { source = "/dev", target = "/dev", type = "devtmpfs", options = [ + "mode=777", + ] } +] + +[base_image] +api_url = "https://registry-1.docker.io" +service = "registry.docker.io" +oauth_url = "https://auth.docker.io/token" +auth_type = "repository" +auth_name = "luxonis/oakapp-base" +image_name = "luxonis/oakapp-base" +image_tag = "1.2.6" diff --git a/m8-controller-box/simple-example/requirements.txt b/m8-controller-box/simple-example/requirements.txt new file mode 100644 index 000000000..2f4dd0fd8 --- /dev/null +++ b/m8-controller-box/simple-example/requirements.txt @@ -0,0 +1,2 @@ +hidapi +git+https://github.com/luxonis/rp2040_u2if.git diff --git a/m8-controller-box/strobe-relay-example/.oakappignore b/m8-controller-box/strobe-relay-example/.oakappignore new file mode 100644 index 000000000..671c0ae52 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/.oakappignore @@ -0,0 +1,33 @@ +# Python virtual environments +venv/ +.venv/ + +# Node.js +# ignore node_modules, it will be reinstalled in the container +node_modules/ + +# Multimedia files +media/ + +# Documentation +README.md + +# VCS +.git/ +.github/ +.gitlab/ + +# The following files are ignored by default +# uncomment a line if you explicitly need it + +# !*.oakapp + +# Python +# !**/.mypy_cache/ +# !**/.ruff_cache/ + +# IDE files +# !**/.idea +# !**/.vscode +# !**/.zed + diff --git a/m8-controller-box/strobe-relay-example/README.md b/m8-controller-box/strobe-relay-example/README.md new file mode 100644 index 000000000..50e4f1594 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/README.md @@ -0,0 +1,47 @@ +# Barcode Detection on Conveyor Belt – Controller Box Strobe Example + +This example demonstrates **real-time barcode detection and decoding** on a conveyor belt using DepthAI cameras. It uses a combination of **pyzbar** and multi-frame validation to ensure robust detection. Additionally, this version includes **temporal smoothing** of bounding boxes to reduce jitter and improve visual tracking. + +> **Note:** This example works only on OAK4 in standalone mode. It is not supported in host-driven peripheral mode because the M8 Controller Box is connected directly to the OAK4 device. + +## Functionality + +* Detects and decodes barcodes from live camera feed. +* Draws **bounding boxes** around detected barcodes, with temporal smoothing over 2–3 frames. +* Highlights **valid barcodes** in green, other detections in red. +* Stops and restarts a conveyor using a simple **state machine** when barcodes are detected. +* Streams live video over HTTP MJPEG for visualization. + +## Recommended Devices + +* **OAK4-CS** – Best for fast-moving conveyor belts, global-shutter color sensor minimizes motion blur. +* **OAK4-S / OAK4-D** – Works with good lighting; best to keep barcodes near optimal focus distance. + +> Fixed-focus cameras may result in inconsistent detection and decoding performance. + +## Demo + +![Demo](media/conveyor_application.gif) + +### Standalone Mode (RVC4 only) + +This app is designed to run entirely on the device. + +1. Install `oakctl` by following the instructions [here](https://docs.luxonis.com/software-v3/oak-apps/oakctl). +2. From this directory, connect to the device and run the app: + +```bash +oakctl connect +oakctl app run . +``` + +### Peripheral Mode (Host + Device) + +This mode is not supported for this example. The M8 Controller Box must be attached directly to the OAK4 device running the app. + +### Configuration + +* `TEMPORAL_SMOOTHING` – Toggle smoothing of bounding boxes. +* `SMOOTHING_ALPHA` – Adjust smoothing weight. +* `STOP_DURATION` – Duration to stop conveyor on barcode detection. +* `COOLDOWN_TIME` – Time before a scanned barcode expires. diff --git a/m8-controller-box/strobe-relay-example/backend-run.sh b/m8-controller-box/strobe-relay-example/backend-run.sh new file mode 100755 index 000000000..e1c1f5c81 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/backend-run.sh @@ -0,0 +1,10 @@ +#!/bin/sh +# run as root inside container +echo "Starting Backend" +# echo "Setting GPIO 32 to 0" +# gpioset 0 32=0 +# echo "Switching USB role" +# echo "host" > /sys/class/usb_role/a600000.ssusb-role-switch/role +# # start main app +# echo "Starting main.py" +exec python3 -u /app/main.py \ No newline at end of file diff --git a/m8-controller-box/strobe-relay-example/main.py b/m8-controller-box/strobe-relay-example/main.py new file mode 100644 index 000000000..77799deff --- /dev/null +++ b/m8-controller-box/strobe-relay-example/main.py @@ -0,0 +1,190 @@ +""" +Main Application +---------------- +• Video stream from DepthAI (HTTP MJPEG) +• Barcode detection pipeline with optional temporal smoothing +• Conveyor + FSYNC control +""" + +import time +import cv2 +import threading +import numpy as np +import depthai as dai +from http.server import BaseHTTPRequestHandler, HTTPServer +from socketserver import ThreadingMixIn + +from utils.barcode_processor import BarcodeProcessor +from utils.box_config import BoxConfig + +# ------------------------------------------------------------ +# CONFIGURATION +# ------------------------------------------------------------ +HTTP_SERVER_PORT = 8083 # MJPEG server port +TEMPORAL_SMOOTHING = True # Enable/disable bounding box smoothing +SMOOTHING_ALPHA = 0.5 # 0.0 = max smoothing, 1.0 = no smoothing +FONT = cv2.FONT_HERSHEY_SIMPLEX +COLOR_VALID = (0, 255, 0) +COLOR_INVALID = (0, 0, 255) + + +# ------------------------------------------------------------ +# MJPEG streaming server +# ------------------------------------------------------------ +class VideoStreamHandler(BaseHTTPRequestHandler): + """HTTP server to stream frames as MJPEG""" + def do_GET(self): + self.send_response(200) + self.send_header( + "Content-type", "multipart/x-mixed-replace; boundary=--jpgboundary" + ) + self.end_headers() + while True: + time.sleep(0.03) + if hasattr(self.server, "frametosend"): + ok, encoded = cv2.imencode(".jpg", self.server.frametosend) + self.wfile.write(b"--jpgboundary\r\n") + self.send_header("Content-type", "image/jpeg") + self.send_header("Content-length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded.tobytes()) + self.wfile.write(b"\r\n") + + +class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): + """Handle MJPEG requests in a separate thread.""" + pass + + +# ------------------------------------------------------------ +# Host node: Barcode detection + conveyor control + frame output +# ------------------------------------------------------------ +class BarcodeHostNode(dai.node.ThreadedHostNode): + def __init__(self, box: BoxConfig, server: ThreadedHTTPServer): + super().__init__() + self.input = self.createInput() + self.input.setBlocking(False) + self.output = self.createOutput() + self.processor = BarcodeProcessor() + self.box = box + self.server = server + + # State machine for conveyor + self.STATE_RUNNING = 0 + self.STATE_STOPPED = 1 + self.STATE_COOLDOWN = 2 + self.state = self.STATE_RUNNING + self.state_timestamp = time.time() + self.STOP_DURATION = 1.5 + self.COOLDOWN_TIME = 2.0 + + # Temporal smoothing storage + self.last_rects = {} # barcode_data -> (x, y, w, h) + + def run(self): + while self.isRunning(): + in_msg = self.input.tryGet() + if in_msg is None: + time.sleep(0.001) + continue + + frame = in_msg.getCvFrame() + + # Detect barcodes + barcodes = self.processor.decode_barcodes(frame) + valid = self.processor.filter_valid_barcodes(barcodes) + + # Print all valid barcodes to console + for data, _ in valid: + print(f"[INFO] Barcode detected: {data}") + + # Draw all detected barcodes with optional temporal smoothing + for bc in barcodes: + data = bc.data.decode("utf-8") + x, y, w, h = bc.rect + + if TEMPORAL_SMOOTHING: + if data in self.last_rects: + lx, ly, lw, lh = self.last_rects[data] + # Weighted average for smooth motion + x = int(lx * (1 - SMOOTHING_ALPHA) + x * SMOOTHING_ALPHA) + y = int(ly * (1 - SMOOTHING_ALPHA) + y * SMOOTHING_ALPHA) + w = int(lw * (1 - SMOOTHING_ALPHA) + w * SMOOTHING_ALPHA) + h = int(lh * (1 - SMOOTHING_ALPHA) + h * SMOOTHING_ALPHA) + self.last_rects[data] = (x, y, w, h) + + # Color green for valid, red for detected but not valid + color = COLOR_VALID if any(v[0] == data for v in valid) else COLOR_INVALID + cv2.rectangle(frame, (x, y), (x + w, y + h), color, 2) + cv2.putText(frame, data, (x, y - 10), FONT, 0.5, color, 2) + + # Conveyor state machine + now = time.time() + if self.state == self.STATE_RUNNING and valid: + self.box.stop_conveyor() + self.state = self.STATE_STOPPED + self.state_timestamp = now + + elif self.state == self.STATE_STOPPED: + if now - self.state_timestamp >= self.STOP_DURATION: + print("[INFO] Restarting conveyor") + self.box.start_conveyor() + self.state = self.STATE_COOLDOWN + self.state_timestamp = now + + elif self.state == self.STATE_COOLDOWN: + if now - self.state_timestamp >= self.COOLDOWN_TIME: + self.state = self.STATE_RUNNING + + # Update MJPEG server frame + self.server.frametosend = frame + + # Optionally send frame forward + out = dai.ImgFrame() + out.setData(frame) + out.setWidth(frame.shape[1]) + out.setHeight(frame.shape[0]) + self.output.send(out) + + +# ------------------------------------------------------------ +# MAIN +# ------------------------------------------------------------ +if __name__ == "__main__": + # Initialize DepthAI device + device = dai.Device() + box = BoxConfig() + box.init_fsync() + platform = device.getPlatform().name + print(f"[INFO] Platform: {platform}") + + # Choose frame type based on platform + frame_type = dai.ImgFrame.Type.BGR888i if platform == "RVC4" else dai.ImgFrame.Type.BGR888p + + # Start MJPEG server + server = ThreadedHTTPServer(("0.0.0.0", HTTP_SERVER_PORT), VideoStreamHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + print(f"[INFO] MJPEG stream available at http://:{HTTP_SERVER_PORT}") + + with dai.Pipeline(device) as pipeline: + # Camera node + cam = pipeline.create(dai.node.Camera).build() + cam.initialControl.setManualExposure(3000, 200) + device.setExternalFrameSyncRole(dai.ExternalFrameSyncRole.MASTER) + + # Request full sensor resolution for best FOV + cam_out = cam.requestOutput((1920, 1080), frame_type, fps=30) + + # Host node for barcode processing + conveyor + barcode_node = pipeline.create(BarcodeHostNode, box, server) + cam_out.link(barcode_node.input) + + # Run the pipeline + pipeline.run() + print("[INFO] Pipeline running...") + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("[INFO] Exiting...") \ No newline at end of file diff --git a/m8-controller-box/strobe-relay-example/media/conveyor_application.gif b/m8-controller-box/strobe-relay-example/media/conveyor_application.gif new file mode 100644 index 000000000..609a15947 Binary files /dev/null and b/m8-controller-box/strobe-relay-example/media/conveyor_application.gif differ diff --git a/m8-controller-box/strobe-relay-example/oakapp.toml b/m8-controller-box/strobe-relay-example/oakapp.toml new file mode 100644 index 000000000..9f012b593 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/oakapp.toml @@ -0,0 +1,39 @@ +identifier = "com.example.object-detection.fsync-relay-barcode" +app_version = "1.0.0" + +allowed_devices = [{ allow = true, access = "rwm" }] + +# Install dependencies and run system-level commands +prepare_container = [ + # system packages + { type = "RUN", command = "apt-get update" }, + { type = "RUN", command = "apt-get install -y python3 python3-pip libzbar0 libzbar-dev libhidapi-hidraw0 libhidapi-libusb0 git gpiod" }, + + # python dependencies + { type = "COPY", source = "requirements.txt", target = "requirements.txt" }, + { type = "RUN", command = "pip3 install -r /app/requirements.txt --break-system-packages" }, +] + +build_steps = [ + "mkdir -p /etc/service/backend", + "cp /app/backend-run.sh /etc/service/backend/run", + "chmod +x /etc/service/backend/run", + "chmod +x /app/backend-run.sh", +] + +additional_mounts = [ + { source = "/dev", target = "/dev", type = "devtmpfs", options = [ + "mode=777", + ] }, +] + +entrypoint = ["bash", "-c", "/usr/bin/runsvdir -P /etc/service"] + +[base_image] +api_url = "https://registry-1.docker.io" +service = "registry.docker.io" +oauth_url = "https://auth.docker.io/token" +auth_type = "repository" +auth_name = "luxonis/oakapp-base" +image_name = "luxonis/oakapp-base" +image_tag = "1.2.6" diff --git a/m8-controller-box/strobe-relay-example/requirements.txt b/m8-controller-box/strobe-relay-example/requirements.txt new file mode 100644 index 000000000..594bb4066 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/requirements.txt @@ -0,0 +1,7 @@ +hidapi +git+https://github.com/luxonis/rp2040_u2if.git +depthai +numpy +opencv-python-headless +pyzbar +Pillow \ No newline at end of file diff --git a/m8-controller-box/strobe-relay-example/utils/barcode_processor.py b/m8-controller-box/strobe-relay-example/utils/barcode_processor.py new file mode 100644 index 000000000..5875aa409 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/utils/barcode_processor.py @@ -0,0 +1,100 @@ +""" +Barcode Processing Module +------------------------ + +Handles: +• Barcode detection using pyzbar +• Debounce (avoid repeated triggers for same barcode) +• Multi-frame confirmation +""" + +import time +import cv2 +from pyzbar.pyzbar import decode + + +class BarcodeProcessor: + def __init__(self): + # ---------------------------------------------------- + # Debounce settings + # ---------------------------------------------------- + self.cooldown_seconds = 3.0 # ignore same barcode for this time + self.last_seen = {} # {barcode_data: timestamp} + + # ---------------------------------------------------- + # Multi-frame confirmation + # ---------------------------------------------------- + self.required_frames = 3 + self.frame_counts = {} # {barcode_data: count} + + def preprocess(self, frame): + """ + Improve detection robustness + """ + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + + # Histogram equalization improves contrast + gray = cv2.equalizeHist(gray) + + return gray + + def decode_barcodes(self, frame): + """ + Detect and decode barcodes + """ + processed = self.preprocess(frame) + return decode(processed) + + def filter_valid_barcodes(self, barcodes): + """ + Apply: + • Multi-frame confirmation + • Debounce logic + """ + valid = [] + now = time.time() + + for bc in barcodes: + data = bc.data.decode("utf-8") + + # ------------------------------------------------ + # Multi-frame confirmation + # ------------------------------------------------ + self.frame_counts[data] = self.frame_counts.get(data, 0) + 1 + + if self.frame_counts[data] < self.required_frames: + continue + + # ------------------------------------------------ + # Debounce (cooldown) + # ------------------------------------------------ + last_time = self.last_seen.get(data, 0) + + if now - last_time < self.cooldown_seconds: + continue + + # Accept barcode + self.last_seen[data] = now + self.frame_counts[data] = 0 + valid.append((data, bc)) + + return valid + + def draw_barcodes(self, frame, barcodes): + """ + Draw bounding boxes + text + """ + for data, bc in barcodes: + x, y, w, h = bc.rect + + cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2) + + cv2.putText( + frame, + data, + (x, y - 10), + cv2.FONT_HERSHEY_SIMPLEX, + 0.5, + (0, 255, 0), + 2, + ) \ No newline at end of file diff --git a/m8-controller-box/strobe-relay-example/utils/box_config.py b/m8-controller-box/strobe-relay-example/utils/box_config.py new file mode 100644 index 000000000..773ed7f43 --- /dev/null +++ b/m8-controller-box/strobe-relay-example/utils/box_config.py @@ -0,0 +1,71 @@ +""" +Box Configuration Module +------------------------ + +Handles: +• Initialization of ControllerBox (single instance) +• Relay control (relay 1) +• FSYNC controller setup +• LED indication +""" + +import time +from luxonis_u2if import ControllerBox + + +class BoxConfig: + """ + Singleton-style ControllerBox manager + Provides: + • Relay control + • FSYNC initialization + """ + + def __init__(self): + # ---------------------------------------------------- + # Connect to ControllerBox (only once!) + # ---------------------------------------------------- + self.box = ControllerBox() + + # ---------------------------------------------------- + # Initialize hardware + # ---------------------------------------------------- + self.box.relay_init() + self.box.led_init() + + # Relay configuration + self.relay_id = 1 + + # FSYNC output selection + self.fsync_out = ControllerBox.FsyncOutput.ISOLATED_STROBE + + # -------------------------------------------------------- + # Relay control + # -------------------------------------------------------- + def stop_conveyor(self): + """ + Activate relay → stop conveyor + """ + self.box.led_on(0) + self.box.relay_reset(self.relay_id) + + def start_conveyor(self): + """ + Deactivate relay → start conveyor + """ + self.box.led_off(0) + self.box.relay_set(self.relay_id) + + # -------------------------------------------------------- + # FSYNC control + # -------------------------------------------------------- + def init_fsync(self, mode=ControllerBox.FsyncMode.SLAVE, frequency=0.1, polarity=True, duty_cycle=50.0): + """ + Initialize FSYNC controller. + Defaults to SLAVE mode with low frequency. + """ + self.box.fsync_controller_init() + self.box.fsync_controller_set_mode(mode) + self.box.fsync_controller_set_frequency(frequency) + self.box.fsync_controller_set_polarity(polarity, self.fsync_out) + self.box.fsync_controller_set_duty_cycle(duty_cycle, self.fsync_out) \ No newline at end of file