Troubleshooting Long-Term Data Collection with Lepton 3.1R and pureThermal3 on Raspberry Pi 4B

[AyomiSakai]

I connected the Lepton 3.1R and pureThermal3 to a Raspberry Pi 4B via a USB cable to create a thermal camera. However, it is currently unusable because it cannot collect data over extended periods. Therefore, I would like to inquire about issues related to the pureThermal3 code.

I developed code to save PNG and CSV files by referencing the official groupgets code here, specifically the uvc-radiometry.py script. I saved this code as a script and configured another program to execute this script four times per minute. Initially, PNG files and data were saved successfully, but after running for about ten minutes, the Lepton stopped transmitting data and receiving commands (such as FFC).

When checking with lsusb, the USB connection appears to be intact, and video capture with OpenCV operates for extended periods. I believe that the frequent execution of the script to save PNG and CSV files, which repeatedly connects and disconnects the pureThermal, caused this error. However, I have not yet determined where to make specific corrections.

What modifications should I make to enable long-term data collection?

The code for saving PNG and CSV files is as follows:
#!/usr/bin/env python

-- coding: utf-8 --

from uvctypes import *
from ctypes import *
import time
import cv2
import numpy as np
from queue import Queue
import platform
import csv
import os
from datetime import datetime
import sys

Set buffer size

BUF_SIZE = 2
q = Queue(BUF_SIZE)

Set device Vendor ID and Product ID

Replace these values with those of your camera

PT_USB_VID = 0x1e4e # Example: Camera Vendor ID
PT_USB_PID = 0x0100 # Example: Camera Product ID

Load libuvc (modify the path if necessary)

try:
libuvc = CDLL('libuvc.so')
except OSError as e:
print(f"Failed to load libuvc: {e}")
sys.exit(1)

def py_frame_callback(frame, userptr):
"""
Frame callback function.
Called when a new frame is received from the camera.
"""
array_pointer = cast(frame.contents.data, POINTER(c_uint16 * (frame.contents.width * frame.contents.height)))
data = np.frombuffer(
array_pointer.contents, dtype=np.dtype(np.uint16)
).reshape(
frame.contents.height, frame.contents.width
) # No data copy

# Verify data size
if frame.contents.data_bytes != (2 * frame.contents.width * frame.contents.height):
    print("Unexpected frame size. Skipping frame.")
    return

# Add data to the queue if it's not full
if not q.full():
    q.put(data)

Register the callback function

PTR_PY_FRAME_CALLBACK = CFUNCTYPE(None, POINTER(uvc_frame), c_void_p)(py_frame_callback)

def ktoc(val):
"""
Convert temperature data from Kelvin to Celsius.
"""
return (val - 27315) / 100.0

def raw_to_8bit(data):
"""
Function to convert 16-bit raw data to 8-bit.
"""
cv2.normalize(data, data, 0, 65535, cv2.NORM_MINMAX)
np.right_shift(data, 8, data)
return np.uint8(data)

def main():
# Set the save directory
save_directory = "/home/*******"

# Create the directory if it does not exist
if not os.path.exists(save_directory):
    try:
        os.makedirs(save_directory)
        print(f"Created directory: {save_directory}")
    except Exception as e:
        print(f"Failed to create directory {save_directory}: {e}")
        sys.exit(1)

ctx = POINTER(uvc_context)()
dev = POINTER(uvc_device)()
devh = POINTER(uvc_device_handle)()
ctrl = uvc_stream_ctrl()

res = libuvc.uvc_init(byref(ctx), 0)
if res < 0:
    print("uvc_init error")
    sys.exit(1)

try:
    res = libuvc.uvc_find_device(ctx, byref(dev), PT_USB_VID, PT_USB_PID, 0)
    if res < 0:
        print("uvc_find_device error")
        sys.exit(1)

    try:
        res = libuvc.uvc_open(dev, byref(devh))
        if res < 0:
            print("uvc_open error")
            sys.exit(1)

        # Display device information (comment out if not needed)
        # print_device_info(devh)
        # print_device_formats(devh)

        frame_formats = uvc_get_frame_formats_by_guid(devh, VS_FMT_GUID_Y16)
        if len(frame_formats) == 0:
            print("Device does not support Y16 format.")
            sys.exit(1)

        libuvc.uvc_get_stream_ctrl_format_size(
            devh,
            byref(ctrl),
            UVC_FRAME_FORMAT_Y16,
            frame_formats[0].wWidth,
            frame_formats[0].wHeight,
            int(1e7 / frame_formats[0].dwDefaultFrameInterval),
        )

        res = libuvc.uvc_start_streaming(devh, byref(ctrl), PTR_PY_FRAME_CALLBACK, None, 0)
        if res < 0:
            print(f"uvc_start_streaming failed: {res}")
            sys.exit(1)

        try:
            # Get the current date and time to generate filenames
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            csv_filename = os.path.join(save_directory, f"temperature_{timestamp}.csv")
            png_filename = os.path.join(save_directory, f"thermal_image_{timestamp}.png")

            # Retrieve and process one frame
            data = q.get(True, 6)  # Set timeout to 6 seconds
            if data is not None:
                # Convert temperature data to Celsius
                temp_data = ktoc(data)

                # Save temperature data to CSV file
                try:
                    np.savetxt(csv_filename, temp_data, delimiter=",", fmt="%.2f")
                    print(f"Saved temperature data to {csv_filename}")
                except Exception as e:
                    print(f"Failed to save CSV file: {e}")

                # Convert image data to 8-bit
                try:
                    data_8bit = raw_to_8bit(data)
                except Exception as e:
                    print(f"Failed to convert data to 8-bit: {e}")
                    data_8bit = None

                if data_8bit is not None:
                    # Apply rainbow color scale
                    color_image = cv2.applyColorMap(data_8bit, cv2.COLORMAP_JET)

                    # Save the image as a PNG file
                    try:
                        cv2.imwrite(png_filename, color_image)
                        print(f"Saved thermal image to {png_filename}")
                    except Exception as e:
                        print(f"Failed to save PNG file: {e}")
                else:
                    print("8-bit image data is invalid. Skipping image save.")
            else:
                print("No data received from camera.")

        finally:
            # Stop streaming
            libuvc.uvc_stop_streaming(devh)
            # print("Streaming stopped.")

        # print("Process completed successfully.")
    finally:
        # Release the device reference
        libuvc.uvc_unref_device(dev)
finally:
    # Exit the UVC context
    libuvc.uvc_exit(ctx)

if name == 'main':
main()