Kutluhan Aktar
Published © CC BY

Web-Enabled ML Mask Detection Robot Fines w/ Penalty Receipt

A prototype to minimize the number of staffs having to interact w/ people to notify them wearing masks live streaming while operating.

ExpertFull instructions provided8 hours6,073

Things used in this project

Hardware components

Raspberry Pi 3 Model B+
Raspberry Pi 3 Model B+
Raspberry Pi 3 Model B+ or 4
×1
Raspberry Pi 4 Model B
Raspberry Pi 4 Model B
Raspberry Pi 3 Model B+ or 4
×1
DFRobot HUSKYLENS with Silicone Case
×1
DFRobot Tiny (Embedded) Thermal Printer - TTL Serial
×1
DFRobot Black Gladiator - Tracked Robot Chassis
×1
USB Webcam
×1
SparkFun L298N Motor Driver Module
×1
12V External Battery
×1
MB102 Power Supply Module
×1
Xiaomi 20000 mAh 3 Pro Type-C Powerbank
×1
Jumper wires (generic)
Jumper wires (generic)
×1

Software apps and online services

Raspbian
Raspberry Pi Raspbian
Raspberry Pi Thonny
Visual Studio 2017
Microsoft Visual Studio 2017
Or any other Editor for PHP coding

Hand tools and fabrication machines

Soldering iron (generic)
Soldering iron (generic)
Solder Wire, Lead Free
Solder Wire, Lead Free
Hot glue gun (generic)
Hot glue gun (generic)

Story

Read more

Custom parts and enclosures

Autonomous_Mask_Detection_Robot.zip

Mask_Detection_Robot_Dashboard.zip

mask_recognition_data_set.zip

General Instruction Set.pdf

HUSKYLENSPython-master.zip

HUSKYLENS Uploader-V2.1.zip

HUSKYLENSWithModelV0.5.1Norm.kfpkg

Schematics

Schematic-1

Schematic-2

Code

mask_detection_robot.py

Python
# Web-enabled ML Mask Detection Robot Fines for No Mask w/ Penalty Receipt

# Raspberry Pi 3 Model B+ or 4

# By Kutluhan Aktar

# A prototype to minimize the number of staffs having to interact w/ people
# to notify them wearing masks live streaming while operating.

# For more information:
# https://www.theamplituhedron.com/projects/Web-enabled-ML-Mask-Detection-Robot-Fines-for-No-Mask-w-Penalty-Receipt

from huskylib import HuskyLensLibrary
import json
from time import sleep
import datetime
import string
import random
from subprocess import call
import requests
from escpos import *

# Create the Mask Detection Robot class with the required settings:
class Mask_Detection_Robot:
    def __init__(self, server, file_location):
        # Define the IP Address and the file location:
        self.server = server
        self.file_location = file_location
        # Define HusklyLens settings
        self.husky_lens = HuskyLensLibrary("I2C", "", address=0x32)
        self.husky_lens_ID = 0
        # Define the case code - unique for each case.
        self.case_code = "default"
        # Test the communication with the HuskyLens.
        print("First request testing: {}".format(self.husky_lens.knock()))
    
    # Decode the data generated by the HuskyLens.
    def decodeHuskyLens(self, obj):
        count=1
        if(type(obj)==list):
            for i in obj:
                #print("\t " + ("BLOCK_" if i.type=="BLOCK" else "ARROW_") + str(count) + " : " + json.dumps(i.__dict__))
                self.husky_lens_ID = json.loads(json.dumps(i.__dict__))["ID"]
                count+=1
        else:
            #print("\t " + ("BLOCK_" if obj.type=="BLOCK" else "ARROW_") + str(count) + " : " + json.dumps(obj.__dict__))
            self.husky_lens_ID = json.loads(json.dumps(obj.__dict__))["ID"]
            
    # Generate a unique case code for each detected people without a mask.
    def generate_unique_case_code(self, length):
        letters_and_digits = string.ascii_letters + string.digits
        self.case_code = ''.join(random.choice(letters_and_digits) for i in range(length))
        print("\nCase Code Generated => " + self.case_code)
    
    # Capture people without a mask when detected by the Huskylens.
    def capture_unmasked(self, width, height, case_code, file_path):
        command_capture = "fswebcam -D 2 -S 20 -r " + width + "x" + height + " " + file_path + case_code + ".jpg"
        command_stop_motion = "sudo service motion stop"
        command_start_motion = "sudo service motion start"
        # Take a picture after interrupting the motion module.
        print("\nStatus => Motion Module Interrupted!")
        call([command_stop_motion], shell=True)
        sleep(10)
        print("\nStatus => Fswebcam Module Activated!\n")
        call([command_capture], shell=True)
        sleep(5)
        print("\nStatus => Motion Module Restarted!\n")
        call([command_start_motion], shell=True)
        sleep(5)
    
    # Send the recently captured pictures of people without a mask to the server.
    def send_captured_img_to_server(self, case_code, file_path):
        url_path = "http://" + self.server + "/Mask_Detection_Robot_Dashboard/captured/"
        captured_image_path = file_path + case_code + ".jpg"
        files = {'captured_image': open(captured_image_path, 'rb')}
        # Make an HTTP Post Request to the server to send the captured image.
        request = requests.post(url_path, files=files)
        print("\nStatus => Recently Captured Image Transferred!")
        # Print the response from the server.
        print("\nServer: " + request.text + "\n")
        
    # Via the Thermal Printer, print the fine receipt when detecting people without a mask.
    def print_fine_receipt(self, case_code, fine, due):
        print("\nStatus => Printer Working!")
        # Define character design and font sizes for each line.
        command_thermal_printer = []
        command_thermal_printer.append("sudo chmod 666 /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1B\x45\x01' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x42\x01' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x21\x37' > /dev/usb/lp0")
        command_thermal_printer.append("echo '&&&' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x42\x10' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x21\x24\x0a' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'COVID-19' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x21\x12' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Management' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Violation' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Notice' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x21\x01' > /dev/usb/lp0")
        command_thermal_printer.append("echo '\\nReceipt No:' > /dev/usb/lp0")
        command_thermal_printer.append("echo '" + case_code + "\\n' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Issue Date: " + datetime.datetime.now().strftime('%m-%d-%Y') + "' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Time: " + datetime.datetime.now().strftime('%H:%M:%S') + "' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Amount: " + fine + "' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'Due Date: In " + due + " Days\\n' > /dev/usb/lp0")
        command_thermal_printer.append("printf '\x1D\x21\x09' > /dev/usb/lp0")
        command_thermal_printer.append("echo '(!) You comitted' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'a code' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'violation by not' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'wearing a mask' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'in public.\\n' > /dev/usb/lp0")
        command_thermal_printer.append("echo '(!) To inspect the' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'picture showing' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'you not wearing' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'a mask in public' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'and pay the penalty' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'enter your' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'receipt number' > /dev/usb/lp0")
        command_thermal_printer.append("echo 'to this page:\\n' > /dev/usb/lp0")
        
        # Print each line via the serial port.
        for i in range(len(command_thermal_printer)):
            call([command_thermal_printer[i]], shell=True)
            sleep(0.5)
        # Print QR Code w/ Settings
        thermal_printer = printer.File("/dev/usb/lp0")
        thermal_printer.qr("http://" + self.server + "/Mask_Detection_Robot_Dashboard/Payments/?q=" + case_code, size=4, model=2)
        thermal_printer.cut()
        print("\nStatus => Printer Printed the Receipt!\n\n")
          
    def MASK_DETECTION(self):
        # Get the recently read block from the HuskyLens to detect the object ID.
        self.decodeHuskyLens(self.husky_lens.blocks())
        if(self.husky_lens_ID == 1):
            print("ID = " + str(self.husky_lens_ID) + " Status => Masked")
        elif(self.husky_lens_ID == 2):
            print("ID = " + str(self.husky_lens_ID) + " Status => UnMasked")
            # Generate the case code.
            self.generate_unique_case_code(15)
            # Capture people without a mask.
            self.capture_unmasked("640", "480", self.case_code, self.file_location)
            # Send the captured image to the server.
            self.send_captured_img_to_server(self.case_code, self.file_location)
            # Print the fine receipt with the penalty.
            self.print_fine_receipt(self.case_code, "$50", "3")
        elif(self.husky_lens_ID == 3):
            print("ID = " + str(self.husky_lens_ID) + " Status => Default")
        

# Define a new class object named 'robot':
robot = Mask_Detection_Robot("192.168.1.24", "/home/pi/Autonomous_Mask_Detection_Robot/captured/")

while True:
    # Get blocks from the HuskyLens:
    robot.MASK_DETECTION()
    sleep(2)

chassis_controls.py

Python
import argparse
import RPi.GPIO as GPIO
from time import sleep

# Define L298N pins.
in_1_1 = 20
in_1_2 = 21
en_1 = 6
in_2_1 = 19
in_2_2 = 26
en_2 = 13

# GPIO Settings
GPIO.setmode(GPIO.BCM)

GPIO.setup(in_1_1, GPIO.OUT)
GPIO.setup(in_1_2, GPIO.OUT)
GPIO.setup(en_1, GPIO.OUT)
GPIO.setup(in_2_1, GPIO.OUT)
GPIO.setup(in_2_2, GPIO.OUT)
GPIO.setup(en_2, GPIO.OUT)

GPIO.output(in_1_1, GPIO.LOW)
GPIO.output(in_1_2, GPIO.LOW)
GPIO.output(in_2_1, GPIO.LOW)
GPIO.output(in_2_2, GPIO.LOW)

s_1 = GPIO.PWM(en_1, 100)
s_1.start(50)
s_2 = GPIO.PWM(en_2, 100)
s_2.start(50)


# If the file is not imported:
if __name__ == '__main__': 
   parser = argparse.ArgumentParser()
   parser.add_argument("--direction", required=True, help="define the direction of the robot chassis")
   parser.add_argument("--speed", help="define the speed of the robot chassis")
   args = parser.parse_args()
   # Mandatory Direction Controls:
   if(args.direction == "forward"):
       GPIO.output(in_1_1, GPIO.LOW)
       GPIO.output(in_1_2, GPIO.HIGH)
       GPIO.output(in_2_1, GPIO.HIGH)
       GPIO.output(in_2_2, GPIO.LOW)
       print("Robot => Going Forward!")
       sleep(1)
   elif(args.direction == "left"):
       GPIO.output(in_1_1, GPIO.LOW)
       GPIO.output(in_1_2, GPIO.HIGH)
       GPIO.output(in_2_1, GPIO.LOW)
       GPIO.output(in_2_2, GPIO.LOW)       
       print("Robot => Going Left!")
       sleep(1)
   elif(args.direction == "right"):
       GPIO.output(in_1_1, GPIO.LOW)
       GPIO.output(in_1_2, GPIO.LOW)
       GPIO.output(in_2_1, GPIO.HIGH)
       GPIO.output(in_2_2, GPIO.LOW)
       print("Robot => Going Right!")
       sleep(1)
   elif(args.direction == "backward"):
       GPIO.output(in_1_1, GPIO.HIGH)
       GPIO.output(in_1_2, GPIO.LOW)
       GPIO.output(in_2_1, GPIO.LOW)
       GPIO.output(in_2_2, GPIO.HIGH)
       print("Robot => Going Backward!")
       sleep(1)
   else:
       print("Direction Value => Error!")
   # Optional Speed Controls:
   if args.speed:
       if(args.speed == "low"):
           s_1.ChangeDutyCycle(50)
           s_2.ChangeDutyCycle(50)
           print("Robot => Low Speed!")
           sleep(1)
       elif(args.speed == "moderate"):
           s_1.ChangeDutyCycle(75)
           s_2.ChangeDutyCycle(75)                
           print("Robot => Moderate Speed!")
           sleep(1)
       elif(args.speed == "high"):
           s_1.ChangeDutyCycle(90)
           s_2.ChangeDutyCycle(90)          
           print("Robot => High Speed!")
           sleep(1)
       else:
           print("Speed Value => Error!")
           
   print("Waiting New Command!")

# Exit and Clear
GPIO.cleanup()

index_php_dashboard.php

PHP
<?php

// Check if there is a new command.
if(isset($_GET["direction"]) && !empty($_GET["direction"])){
	$direction = $_GET["direction"];
	// If transferred, get the optional speed value.
	$speed = (isset($_GET["speed"])) ? $_GET["speed"] : "low";
	// Execute the requested Python file with arguments as the recent commands - direction and speed.
	exec("python /home/pi/Autonomous_Mask_Detection_Robot/chassis_controls.py --direction='".$direction."' --speed='".$speed."'");
	echo "Commands =><br><br>Direction: $direction <br><br>Speed: $speed";
	// Close
	exit();
}

?>

<!DOCTYPE html>
<html>
<head>
 <title>Mask Detection Robot Dashboard</title>
 
 <meta charset="UTF-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 
 <!--link to favicon-->
 <link rel="icon" type="image/png" sizes="36x36" href="icon.png">

 <link rel="stylesheet" type="text/css" href="index.css"></link>
 
 <link rel="preconnect" href="https://fonts.gstatic.com">
 <link href="https://fonts.googleapis.com/css2?family=Stalinist+One&display=swap" rel="stylesheet">

 <!-- link to FontAwesome-->
<link rel="stylesheet" href="https://use.fontawesome.com/releases/v5.14.0/css/all.css">
 
 <!--link to jQuery script-->
 <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js"></script>

</head>
<body>
<?php ini_set('display_errors',1);?> 
<h1><i class="fas fa-head-side-mask"></i> Mask Detection Robot Dashboard</h1>
<div class="container">
<div class="stream">
<h2>Stream</h2>
<!-- Change the address with your Raspberry Pi IP Address. -->
<iframe src="http://192.168.1.24:8081" title="Mask Detection Robot Live Stream"></iframe>
</div>

<div class="controls">
<form class="submit">
<fieldset>
<legend>Controls</legend>
<br>
<section>
<label><input type="radio" name="direction" value="forward" /><span class="mark"></span> F</label>
<label><input type="radio" name="direction" value="right" /><span class="mark"></span> R</label>
<label><input type="radio" name="direction" value="left" /><span class="mark"></span> L</label>
<label><input type="radio" name="direction" value="backward" /><span class="mark"></span> B</label>
<br><br>
</section>
<br><br>
<select name="speed">
<option value="low">LOW</option>
<option value="moderate">MODERATE</option>
<option value="high">HIGH</option>
</select>
<option>
</fieldset>
</form>
</div>

</div>
<br><br>

<script src="index.js" type="text/javascript"></script>

</body>
</html>

index_js_dashboard.js

JavaScript
$(".controls").on("input", ".submit", function(){
	// Get the recent command.
	var direction = document.querySelector('input[name="direction"]:checked').value;
	var speed = document.querySelector('select[name="speed"]').value;
	// Send the recent command to the page to execute related Python script.
	$.ajax({
		url: "?direction=" + direction + "&speed=" + speed,
	    type: "GET",
	    //success: () => {alert("D: " + direction + "\nS: " + speed);}
	});
});

index_captured.php

PHP
<?php

// If the captured images of people without a mask are transferred from the  Mask Detection Robot (Raspberry Pi):
if(!empty($_FILES["captured_image"]['name'])){
	// Image File:
	$captured_image_properties = array(
	    "name" => $_FILES["captured_image"]["name"],
	    "tmp_name" => $_FILES["captured_image"]["tmp_name"],
		"size" => $_FILES["captured_image"]["size"],
		"extension" => pathinfo($_FILES["captured_image"]["name"], PATHINFO_EXTENSION)
	);
	
    // Check whether the uploaded file extensions are in allowed formats.
	$allowed_formats = array('jpg', 'png');
	if(!in_array($captured_image_properties["extension"], $allowed_formats)){
		echo 'FILE => File Format Not Allowed!';
	}else{
		// Check whether the uploaded file sizes exceed the data limit - 5MB.
		if($captured_image_properties["size"] > 5000000){
			echo "FILE => File size cannot exceed 5MB!";
		}else{
			// Save the uploaded image.
			move_uploaded_file($captured_image_properties["tmp_name"], $captured_image_properties["name"]);
			echo "FILE => Saved Successfully!";
		}
	}
}

?>

index_payments.php

PHP
<!DOCTYPE html>
<html>
<head>
 <title>Payments</title>
 
 <meta charset="UTF-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 
 <!--link to favicon-->
 <link rel="icon" type="image/png" sizes="36x36" href="../icon.png">
 
 <style>
     html{background-color:#2E3033;}
	 h2{color:#F3D060;text-align:center;}
	 img{display:block;width:80%;height:auto;margin:auto;border:2px solid #F3D060;}
	 button{display:block;margin:auto;width:50%;background-color:#A5282C;color:white;border-radius:20px;border:3px solid white;cursor:pointer;font-weight:bold;font-size:20px;margin-bottom:20px;}
	 button:hover{background-color:#EE7762;}
 </style>

</head>
<body> 

<?php

// If the receipt number is correct, display the captured image, the penalty, and payment options.
if(isset($_GET["q"]) && !empty($_GET["q"])){
	if(file_exists("../captured/".$_GET["q"].".jpg")){
		echo '
		       <h2>Receipt No: '.$_GET["q"].'</h2>
			   <h2>Amount: $50</h2>
			   <button>PayPal</button>
			   <button>Bank Transfer</button>
			   <img src="../captured/'.$_GET["q"].'.jpg" />
	    ';
	}else{
		echo "<h2>Receipt No: Not Found!</h2>";
	}
}

?>

</body>
</html>

Credits

Kutluhan Aktar

Kutluhan Aktar

79 projects • 291 followers
Self-Taught Full-Stack Developer | @EdgeImpulse Ambassador | Maker | Independent Researcher

Comments