mirror of
https://github.com/StefBuwalda/ProjectIOT.git
synced 2025-10-30 11:19:57 +00:00
88 lines
2.2 KiB
Python
88 lines
2.2 KiB
Python
from flask import Flask, request, jsonify
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_migrate import Migrate
|
|
from datetime import datetime
|
|
from pyplatex import ANPR
|
|
from ultralytics.nn.tasks import DetectionModel
|
|
import os
|
|
import torch
|
|
import asyncio
|
|
|
|
torch.serialization.add_safe_globals({"DetectionModel": DetectionModel})
|
|
|
|
# Web Server
|
|
app = Flask(__name__)
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///data.sqlite"
|
|
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
|
app.config["SECRET_KEY"] = "bvjchsygvduycgsyugc"
|
|
|
|
# ORM
|
|
db = SQLAlchemy()
|
|
db.init_app(app)
|
|
|
|
migrate = Migrate(app, db)
|
|
|
|
# Saving images locally
|
|
UPLOAD_FOLDER = "uploads"
|
|
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
|
|
|
|
|
# db classes
|
|
class Plate(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
plate = db.Column(db.String(40), nullable=False)
|
|
|
|
|
|
class LoggedItem(Plate):
|
|
dateLogged = db.Column(db.DateTime, default=datetime.now)
|
|
|
|
|
|
# Default app route
|
|
@app.route("/")
|
|
def home():
|
|
return "Hello, World!"
|
|
|
|
|
|
# API to process vehicle
|
|
@app.route("/api", methods=["POST"])
|
|
def data():
|
|
# Check if image is present
|
|
if "image" not in request.files:
|
|
return jsonify({"error": "No file part"}), 400
|
|
|
|
file = request.files["image"] # get the image from the packet
|
|
|
|
if file.filename == "" or not file.filename:
|
|
return jsonify({"error": "No selected file"}), 400
|
|
|
|
# Check if image ends in .jpg
|
|
if file.filename.lower().endswith(".jpg"):
|
|
filepath = os.path.join(UPLOAD_FOLDER, file.filename)
|
|
file.save(filepath)
|
|
plate = asyncio.run(process_image(filepath))
|
|
return jsonify(
|
|
{
|
|
"message": "File uploaded successfully",
|
|
"filename": file.filename,
|
|
"status": True,
|
|
"anpr": plate,
|
|
}
|
|
)
|
|
|
|
return jsonify({"error": "Only JPEG files allowed"}), 400
|
|
|
|
|
|
async def process_image(file: str):
|
|
anpr = ANPR()
|
|
anpr_info = await anpr.detect(file)
|
|
number_plate = anpr_info["plate_number"]
|
|
if number_plate:
|
|
db.session.add(LoggedItem(plate=number_plate))
|
|
db.session.commit()
|
|
return number_plate
|
|
return "ERROR"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="192.168.137.1", port=2222, debug=True)
|