recorder.service (lepton thermal camera, raspberry pi)

gpio_short_check.py

import RPi.GPIO as GPIO
import time

def are_gpio_shorted(gpio_a = 26, gpio_b = 16, sleep_seconds=1, init_gpio=True, cleanup=True):
    if init_gpio:
        GPIO.setmode(GPIO.BCM)

    GPIO.setup(gpio_a, GPIO.OUT, initial=GPIO.HIGH)
    GPIO.setup(gpio_b, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)

    time.sleep(sleep_seconds)

    val = GPIO.input(gpio_b) == 1

    if cleanup:
        GPIO.cleanup()

    return val

lepton_video_provider.py

import cv2

import numpy as np

from pylepton import Lepton

def lepton_video_provider():
    with Lepton() as l:
        prev_fid = None
        fid = None
        
        while True:
            while fid is None or fid == prev_fid:
                frame, _fid = l.capture()
                fid = _fid
            
            prev_fid = fid

            cv2.normalize(frame, frame, 0, 65535, cv2.NORM_MINMAX) # extend contrast
            np.right_shift(frame, 8, frame)
            frame = np.uint8(frame)
            
            yield frame

pylepton_recorder.py

from time import sleep
from gpio_short_check import are_gpio_shorted
from lepton_video_provider import lepton_video_provider

import os, cv2

if not are_gpio_shorted(gpio_a=26, gpio_b=16, sleep_seconds=0.5):
    print('GPIOs not shorted. Exiting.')
    exit(0)

session_dir = '/home/pi/recordings'
os.makedirs(session_dir, exist_ok=True)

def is_int(s):
    try: 
        int(s)
        return True
    except ValueError:
        return False

sessions = [int(x) for x in os.listdir(session_dir) if is_int(x)]
session_idx = 0 if len(sessions) == 0 else max(sessions) + 1

print(f'Session idx: {session_idx}')

save_dir = os.path.join(session_dir, str(session_idx))
os.makedirs(save_dir)
print(f'Will save session files under {save_dir}')

i = 0
try:
    for frame in lepton_video_provider():
        cv2.imwrite(os.path.join(save_dir, f'frame_{i}.png'), frame)
        i += 1
except Exception as e:
    print(e)
finally:
    print(f'Session {session_idx} finished with {i} saved frames')
    print(f'Directory is at: {save_dir}')
    sleep(10)

recorder.service

[Unit]
Description=Thermal Recorder Service
After=multi-user.target

[Service]
ExecStart=/usr/bin/python3 -u pylepton_recorder.py
WorkingDirectory=/home/pi/code
StandardOutput=inherit
StandardError=inherit
Restart=no
User=pi

[Install]
WantedBy=multi-user.target