PiSpeed Camera

Work Distribution

Chenxi Qian

cq53@cornell.edu

Circuit and Hardware Design

Ultrasonic Speed Testing Design

License Detection and OCR

System Testing

Kuan Lu

kl649@cornell.edu

Computer Vision Design

Database and Website Design

Flask Server Design

System Testing

Appendix

Code A

"""
    MeasureDistance.py
    Created by Chenxi Qian

    measure the distance using HC-SR04 Ultrasonic Ranging Module
"""

import RPi.GPIO as GPIO
import time

def measureDistance(trig, echo):
    # set up GPIO pins for TRIG and ECHO
    TRIG_PIN = trig
    ECHO_PIN = echo
    
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(TRIG_PIN, GPIO.OUT)
    GPIO.setup(ECHO_PIN, GPIO.IN)

    # generate a pulse using TRIG pin
    GPIO.output(TRIG_PIN, False)
    GPIO.output(TRIG_PIN, True)
    time.sleep(0.00001)
    GPIO.output(TRIG_PIN, False)

    pulse_start = 0
    pulse_end = 0

    # set up timer and wait for ECHO response
    while GPIO.input(ECHO_PIN) == 0:
        pulse_start = time.time()
    while GPIO.input(ECHO_PIN) == 1:
        pulse_end = time.time()

    # calculate distance
    pulse_duration = pulse_end - pulse_start
    distance = int(pulse_duration * 17150)

    GPIO.cleanup()

    return time.time(), distance

Code B

"""
    MeasureSpeed.py
    Created by Chenxi Qian

    measure the speed of a moving car using two HC-SR04 Ultrasonic Sensors
"""

import time
from MeasureDistance import measureDistance

def measureSpeed():
    # distance between two Ultrasonic Sensors in cm
    SENSOR_DIS = 57

    # define GPIO pins
    LEFT_TRIG_PIN = 23
    LEFT_ECHO_PIN = 24
    RIGHT_TRIG_PIN = 5
    RIGHT_ECHO_PIN = 6

    # initiate ultrasonic sensors
    # first few measurements would be unstable
    # discard these values
    for _ in range(5):
        measureDistance(LEFT_TRIG_PIN, LEFT_ECHO_PIN)
        time.sleep(0.001)
        measureDistance(RIGHT_TRIG_PIN, RIGHT_ECHO_PIN)

    # calibrate and get initial distance
    _, initDistanceLeft = measureDistance(LEFT_TRIG_PIN, LEFT_ECHO_PIN)
    _, initDistanceRight = measureDistance(RIGHT_TRIG_PIN, RIGHT_ECHO_PIN)

    print("#################### Start Program ####################")
    print("############### Measure Initial Distance ##############")
    print("Left Initial:  ", initDistanceLeft)
    print("Right Initial: ", initDistanceRight)
    print("#######################################################")

    # start timer when the car reached the left sensor
    leftDetected = False
    leftDetectedTime = 0
    while not leftDetected:
        currentTimeLeft, measuredDistanceLeft = measureDistance(LEFT_TRIG_PIN, LEFT_ECHO_PIN)
        time.sleep(0.05)
        if measuredDistanceLeft < initDistanceLeft - 5:
            leftDetected = True
            leftDetectedTime = currentTimeLeft

    print("Left Detected! Time:", leftDetectedTime, "Distance:", measuredDistanceLeft)

    # stop timer when the car reached the right sensor
    rightDetected = False
    rightDetectedTime = 0
    while not rightDetected:
        currentTimeRight, measuredDistanceRight = measureDistance(RIGHT_TRIG_PIN, RIGHT_ECHO_PIN)
        time.sleep(0.05)
        if measuredDistanceRight < initDistanceRight - 5:
            rightDetected = True
            rightDetectedTime = currentTimeRight

    print("Right Detected! Time:", rightDetectedTime, "Distance:", measuredDistanceRight)
    print("#######################################################")
    print("################## Calculating Speed ##################")

    # calculate the speed of the car
    speedCM = round(SENSOR_DIS / (rightDetectedTime - leftDetectedTime), 2)
    speedKM = round(speedCM * 0.036, 2)
    speedMI = round(speedCM * 0.022369, 2)
    print("Measured Speed:", speedCM, "cm/s")
    print("Measured Speed:", speedKM, "km/h")
    print("Measured Speed:", speedMI, "mile/h")
    print("#######################################################")
    return speedMI

Code C

"""
    CaptureLiscense.py
    Created by Chenxi Qian

    capture the car image using Raspberry Pi Camera Rev 1.3
"""

from picamera.array import PiRGBArray
from picamera import PiCamera
import time
import cv2

def captureLicense():
    # capture an image using Raspberry Pi Camera
    camera = PiCamera()
    rawCapture = PiRGBArray(camera)

    time.sleep(0.1)

    camera.capture(rawCapture, format="bgr")
    capturedImage = rawCapture.array

    # prepare image file name
    # all the image would be saved in the /ProcessImage directory
    style = "%Y-%m-%d_%H:%M:%S"
    captureTime = str(time.strftime(style, time.localtime(time.time())))
    imagePath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_0Original.jpg"

    # save the image
    cv2.imwrite(imagePath, capturedImage)

    return captureTime, imagePath

Code D

"""
    LicenseDetect.py
    Created by Chenxi Qian and Kuan Lu

    detect the license plate and crop corresponding area as output
"""

import cv2
import numpy as np
import imutils

### import used for test code ###
import glob
from LicenseOCR import licenseOCR
#################################

# helper sort function
# given four (x, y) coordinate points
# sort four points in the following order
# A -> B -> C -> D
# ------------------
# ----A--------C----
# ----|        |----
# ----|        |----
# ----B--------D----
# ------------------
def sort(location):
    ySorted = sorted(location, key=(lambda x: x[1]))
    xSorted0 = sorted(ySorted[0:2], key=(lambda x: x[0]))
    xSorted1 = sorted(ySorted[2:4], key=(lambda x: x[0]))
    return xSorted0 + xSorted1

def licenseDetect(captureTime, imagePath):
    # prepare intermediate image file name and path
    originalPath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_0Original.jpg"
    noiseReducedImagePath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_1NoiseReduced.jpg"
    edgedImagePath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_2Edged.jpg"
    transformedPath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_3Transformed.jpg"
    croppedImagePath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_4Cropped.jpg"

    # crop the captured image to reduce computation
    img = cv2.imread(imagePath)[400:1024, :]
    cv2.imwrite(originalPath, img)
    
    # convert color to grayscale and perform noise reduction
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) # convert color
    bfilter = cv2.bilateralFilter(gray, 11, 17, 17) # Noise reduction
    cv2.imwrite(noiseReducedImagePath, bfilter)

    # edge detection using canny filter
    edged = cv2.Canny(bfilter, 30, 200)
    cv2.imwrite(edgedImagePath, edged)

    # find rectangles from the image
    # trying to match the area of a license plate
    keypoints = cv2.findContours(edged.copy(), cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
    contours = imutils.grab_contours(keypoints)
    contours = sorted(contours, key=cv2.contourArea, reverse=True)[:10]

    location = None
    try:
        for contour in contours:
            approx = cv2.approxPolyDP(contour, 10, True)
            if len(approx) == 4:
                location = approx
                break

        # transform the titled rectangle to a vertical horizontal rectangle
        originalPoints = np.float32(sort([location[0][0], location[1][0], location[2][0], location[3][0]]))
        mappedPoints = np.float32([[0, 0], [800, 0], [0, 400], [800, 400]])
        transformMatrix = cv2.getPerspectiveTransform(originalPoints, mappedPoints)
        # transformed = cv2.warpPerspective(img, transformMatrix, (800, 400))[95:305, 0:800]  # parameter used for real road test
        transformed = cv2.warpPerspective(img, transformMatrix, (800, 400))[95:305, 30:770]  # parameter used for lab test environment
        cv2.imwrite(transformedPath, transformed)
        transformedGray = cv2.cvtColor(transformed, cv2.COLOR_BGR2GRAY)
        
        # histogram equalization and final color convert
        equalized = cv2.equalizeHist(transformedGray)
        # _, cropped = cv2.threshold(equalized, 60, 255, cv2.THRESH_BINARY)  # parameter used for real road test
        _, cropped = cv2.threshold(equalized, 40, 255, cv2.THRESH_BINARY)  # parameter used for lab test environment
        cv2.imwrite(croppedImagePath, cropped)
        return captureTime, croppedImagePath
    
    except:
        return

################################# TEST CODE #################################

# for img in glob.glob(r"/home/pi/PiSpeedCamera/PiCameraImage/*.jpg"):
#     capturedTime, croppedImagePath = licenseDetect(img[37:-4], img)
#     licenseOCR(capturedTime, croppedImagePath)

#############################################################################

Code E

"""
    LicenseOCR.py
    Created by Chenxi Qian

    detecting license plate content using Tesseract OCR engine
"""

import os
import re

def licenseOCR(captureTime, imagePath):
    # execute tesseract command
    print("OCR Image", imagePath)
    cmd = "tesseract " + imagePath + " /home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_5OCRResult" + " -l eng -psm 7 >> /dev/null"
    os.system(cmd)

    OCRPath = "/home/pi/PiSpeedCamera/ProcessImage/" + captureTime + "_5OCRResult.txt"
    with open(OCRPath) as f:
        # read the OCR result into one line and remove spaces
        OCRMsg = f.read().replace(" ", "").replace("\n", "").upper()
        
        # only preserve capital letters and digits
        detectedPlate = re.sub(u"([^\u0030-\u0039\u0041-\u005a])", "", OCRMsg)
    
    print("OCR Result:", detectedPlate)
    return detectedPlate

Code F

"""
    AddOCRToDatabase.py
    Created by Chenxi Qian and Kuan Lu

    add speed and plate information into website database
"""

import sqlite3

def addOCRToDatabase(speed, plate):
    def add_data(license, num):
        curs.execute("INSERT INTO DHT_data values(datetime('now'), (?), (?))", (license, num))
        conn.commit()
    
    conn = sqlite3.connect("/home/pi/PiWeb/Sensors_Database/test.db")
    curs = conn.cursor()
    add_data(plate, speed)
    conn.close()

Code G

"""
    main.py
    Created by Chenxi Qian and Kuan Lu

    main function to start the program
    currently needs one start for each speed test
    loop version in future work
"""

from MeasureSpeed import measureSpeed
from CaptureLicense import captureLicense
from LicenseDetect import licenseDetect
from LicenseOCR import licenseOCR
from AddOCRToDatabase import addOCRToDatabase

# SPEED_LIMIT = 20  # speed limit 20 mile/h
SPEED_LIMIT = 0  # debug mode, assuming all cars over speeding

# measure the speed of car
detectedSpeed = measureSpeed()

# if the car is over speeding
# capture the license plate and add to the website database
if detectedSpeed > SPEED_LIMIT:
    captureTime, originalImagePath = captureLicense()
    captureTime, croppedImagePath = licenseDetect(captureTime, originalImagePath)
    detectedPlate = licenseOCR(captureTime, croppedImagePath)
    addOCRToDatabase(detectedSpeed, detectedPlate)

H Install Latest Version of OpenCV on Raspbian Buster with Python 3

# install the latest version of opencv
# should be done within 1 min

$ pip3 install opencv-contrib-python==4.5.3.56 -i https://www.piwheels.org/simple

# install dependencies
# remember to add the "-t buster"

$ sudo apt-get update
$ sudo apt-get install libhdf5-dev -t buster
$ sudo apt-get install libatlas-base-dev -t buster
$ sudo apt-get install libjasper-dev -t buster
$ sudo apt-get install libqt4-test -t buster
$ sudo apt-get install libqtgui4 -t buster
$ sudo apt-get update

# you may need to upgrade the numpy to the latest version

$ pip3 install numpy --upgrade

# to test the installation of the opencv
# import it and use "cv2.__version__" to see the installed version

pi@5725:~ $ python3
Python 3.7.3 (default, Jan 22 2021, 20:04:44)
[GCC 8.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import cv2
>>> cv2.__version__
'4.5.3'
>>> quit()
pi@5725:~ $

I Database of Website

'''
  database.py
  Created by Kuan Lu
  Creat database with SQLite.
'''

import sqlite3 as lite
import sys
con = lite.connect('test.db')
with con: 
  cur = con.cursor()
  cur.execute("DROP TABLE IF EXISTS data")
  cur.execute("CREATE TABLE DHT_data(timestamp DATETIME, license TEXT, speed NUMERIC)")

J Flask Server

'''
  webServer.py
  
  Created by Kuan Lu   
  Build the server with Flask.
'''

from flask import Flask, render_template, request
app = Flask(__name__)

import sqlite3

# Retrieve data from database
def getData():
  conn=sqlite3.connect('/home/pi/PiWeb/Sensors_Database/test.db')
  curs=conn.cursor()
  time = []
  licensePlate = []
  speed = []
  for row in curs.execute("SELECT * FROM DHT_data ORDER BY timestamp DESC LIMIT 5"):
    time.append(str(row[0]))
    licensePlate.append(row[1])
    speed.append(row[2])
  conn.close()
  return time, licensePlate, speed 

# Upload the data to html 
@app.route("/")
def index():
  
  time, licensePlate, speed = getData()
  templateData = {
    'time'    : time,
    'license' : licensePlate,
    'speed'   : speed
  }
  return render_template('index.html', **templateData)


if __name__ == "__main__":
   app.run(host='0.0.0.0', port=80, debug=False)