camera interface

[iot-root]

Related Issue: https://github.com/iot-root/garden-of-eden/issues/8

List devices available: v4l2-ctl --list-devices

List device formats available: v4l2-ctl --list-formats-ext -d <device>

Example capture: sudo fswebcam -d /dev/video0 -r 2500x1900 -S 2 -F 2 /tmp/capture_a.jpg sudo fswebcam -d /dev/video2 -r 2500x1900 -S 2 -F 2 /tmp/capture_b.jpg

Adding udev rules to auto identify and name cameras

Create new rule sudo nano /etc/udev/rules.d/99-usb-cameras.rules

add and save

SUBSYSTEM=="video4linux", ATTRS{idVendor}=="0c45", ATTRS{idProduct}=="6368", KERNELS=="1-1.3", SYMLINK+="camera_1"
SUBSYSTEM=="video4linux", ATTRS{idVendor}=="0c45", ATTRS{idProduct}=="6368", KERNELS=="1-1.4", SYMLINK+="camera_2"

reload udev rules

sudo udevadm control --reload-rules
sudo udevadm trigger

verify with ls -l /dev/camera_1 /dev/camera_2

Should have something like:

lrwxrwxrwx 1 root root 6 Apr  1 13:35 /dev/camera_1 -> video0
lrwxrwxrwx 1 root root 6 Apr  1 13:35 /dev/camera_2 -> video2

troubleshooting usb devices

1. reload: sudo udevadm control --reload-rules
2. test rule for device: udevadm test $(udevadm info -q path -n /dev/video0)
3. decive info: udevadm info -a -n /dev/video0
4. live monitor: sudo udevadm monitor

sudo udevadm control --reload-rules
sudo udevadm trigger /dev/videoX

[iot-root]

Example publish images but nothing working yet

def publish_images(client):

# while True:
    # try:
        # # Define paths for the left and right camera images
        # left_image_path = '/tmp/camera_left.jpg'
        # right_image_path = '/tmp/camera_right.jpg'

        # # Capture images from the cameras
        # subprocess.run(['fswebcam', '-r', '640x480', '--no-banner', left_image_path], check=True)
        # subprocess.run(['fswebcam', '-r', '640x480', '--no-banner', right_image_path], check=True)

        # # Read and publish the left image
        # with open(left_image_path, 'rb') as image_file:
            # left_image = image_file.read()
        # client.publish(BASE_TOPIC+"/camera/left", payload=left_image)

        # # Read and publish the right image
        # with open(right_image_path, 'rb') as image_file:
            # right_image = image_file.read()
        # client.publish(BASE_TOPIC+"/camera/right", payload=right_image)

    # except subprocess.CalledProcessError as e:
        # logger.error(f"Failed to capture images: {e}")
    # except Exception as e:
        # logger.error(f"Failed to read or publish images: {e}")

    # sleep(60*60*24)  # Adjust the sleep duration as needed

def publish_images(client): while True: try:

Capture and publish images for the left and right positions

        # capture_and_publish_image(client,'left')
        # capture_and_publish_image(client,'right')
        with open('/home/gardyn/projects/garden-of-eden/video0.jpg', "rb") as image_file:
            image_payload = image_file.read()

        client.publish(BASE_TOPIC + f"/camera/left", payload=image_payload, qos=1)

    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        # Handle other unexpected errors

    # Adjust the sleep duration as needed, e.g., for a 1-hour interval, use sleep(3600)
    sleep(10)  # Sleep for 1 hour before the next capture cycle

def publish_images(client):

# while True:
    # try:
        # # Define paths for the left and right camera images
        # left_image_path = '/tmp/camera_left.jpg'
        # right_image_path = '/tmp/camera_right.jpg'

        # # Capture images from the cameras
        # subprocess.check_call(['fswebcam', '-r', '640x480', '--no-banner', left_image_path])
        # subprocess.check_call(['fswebcam', '-r', '640x480', '--no-banner', right_image_path])

        # # Read, encode in Base64, and publish the left image
        # with open(left_image_path, 'rb') as image_file:
            # left_image = base64.b64encode(image_file.read()).decode('utf-8')
        # client.publish(BASE_TOPIC + "/camera/left", payload=left_image)

        # # Read, encode in Base64, and publish the right image
        # with open(right_image_path, 'rb') as image_file:
            # right_image = base64.b64encode(image_file.read()).decode('utf-8')
        # client.publish(BASE_TOPIC + "/camera/right", payload=right_image)

    # except subprocess.CalledProcessError as e:
        # logger.error(f"Failed to capture images: {e}")
        # # Handle specific recovery or retry logic here if necessary
    # except IOError as e:
        # logger.error(f"File IO Error: {e}")
        # # Handle file access or permission errors
    # except Exception as e:
        # logger.error(f"Unexpected error: {e}")
        # # Handle other unexpected errors

    # # Adjust the sleep duration as needed, e.g., for a 1-hour interval, use sleep(3600)
    # sleep(3600)  # Sleep for 1 hour before the next capture cycle

[iot-root]

import subprocess
import threading
import logging
import paho.mqtt.client as mqtt
import base64
import json
from time import sleep
from config import USERNAME, PASSWORD, BROKER, PORT, KEEP_ALIVE_INTERVAL, BASE_TOPIC, IDENTIFIER, MODEL, VERSION

# Configure logging
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def send_discovery_messages(client):
    device_info = {
        "identifiers": [IDENTIFIER],
        "name": BASE_TOPIC,
        "manufacturer": "gardyn-of-eden",
        "model": MODEL,
        "sw_version": VERSION,
    }

    TEMP_CONFIG_TOPIC = "homeassistant/camera/gardyn/"+IDENTIFIER+"_cameraright/config"
    temp_config_payload = {
        "name": "Right Camera",
        "unique_id": IDENTIFIER + "cameraright",
        "state_topic": BASE_TOPIC + "/camera/right",
        "device": device_info
    }
    client.publish(TEMP_CONFIG_TOPIC, json.dumps(temp_config_payload), retain=True)

def on_connect(client, userdata, flags, rc, properties=None):
    logger.info(f"Connected with result code {rc}")
    client.subscribe(BASE_TOPIC + "/#")
    send_discovery_messages(client)

def on_message(client, userdata, msg):
    logger.debug(f"Message received on topic {msg.topic}: {msg.payload}")
    # try:
        # payload = msg.payload.decode("utf-8")
    # except UnicodeDecodeError as e:
        # logger.error(f"Error decoding message: {e}. Data may not be text.")
    # except ValueError as e:
        # logger.error(f"ValueError encountered: {e}")

def publish_images(client):
    while True:
        try:
            with open('image.jpg', "rb") as image_file:
                image_payload = image_file.read()
            byteArr = bytearray(image_payload)
            client.publish(BASE_TOPIC + f"/camera/right", payload=byteArr, qos=1)

        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            # Handle other unexpected errors

        # Adjust the sleep duration as needed, e.g., for a 1-hour interval, use sleep(3600)
        sleep(10)  # Sleep for 1 hour before the next capture cycle

if __name__ == "__main__":
    logger.info(f"Connecting to {BROKER} on port {PORT} with keep alive {KEEP_ALIVE_INTERVAL}")
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.on_connect = on_connect
    client.on_message = on_message
    client.username_pw_set(USERNAME, PASSWORD)
    client.connect(BROKER, PORT, KEEP_ALIVE_INTERVAL)

    publish_images_thread = threading.Thread(target=publish_images, args=(client,))
    publish_images_thread.daemon = True
    publish_images_thread.start()```

[iot-root]

#!/bin/bash
# Usage: script_name resolution filename
# Example: ./capture_image.sh 640x480 image.jpg
# The '-S 2' will skip the first two frames, allows for adjusting to the lighting
# The '-F 2' captures 2 frames to reduce noise
#RESOLUTION="640x480"
RESOLUTION="2560x1920"

#v4l2-ctl -d /dev/camera_1 --set-input=0
fswebcam -d "/dev/camera_1" -r $RESOLUTION -S 2 -F 2 "/tmp/cam_1.jpg"
#sleep 2
#v4l2-ctl -d /dev/camera_2 --set-input=0
fswebcam -d "/dev/camera_2" -r $RESOLUTION -S 2 -F 2 "/tmp/cam_2.jpg"

[iot-root]

https://www.howtoraspberry.com/2022/02/real-time-streaming-protocol-and-the-raspberry-pi-video-for-linux-2/