123123123

AT_frontend/app.py

@@ -1,13 +1,18 @@
 from flask import Flask, render_template, session, redirect, url_for, session
 from flask_wtf import FlaskForm
-from wtforms import (StringField, BooleanField,
+from wtforms import (
-                                  RadioField, SelectField,
+    StringField,
-                                  TextAreaField, SubmitField)
+    BooleanField,
+    RadioField,
+    SelectField,
+    TextAreaField,
+    SubmitField,
+)
 from wtforms.validators import DataRequired
 
 app = Flask(__name__)
 
-app.config['SECRET_KEY'] = 'mijngeheimesleutel'
+app.config["SECRET_KEY"] = "mijngeheimesleutel"
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     app.run(debug=True)

application/__init__.py

@@ -0,0 +1,14 @@
+from flask import Flask
+from flask_migrate import Migrate
+from flask_sqlalchemy import SQLAlchemy
+
+# 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(app)
+
+migrate = Migrate(app, db)

migrations/versions/ac8b1871be3d_.py

@@ -1,8 +1,8 @@
 """empty message
 
-Revision ID: 7a26306b04df
+Revision ID: ac8b1871be3d
 Revises: 
-Create Date: 2025-04-10 20:56:03.819230
+Create Date: 2025-04-22 22:32:26.012696
 
 """
 from alembic import op
@@ -10,7 +10,7 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = '7a26306b04df'
+revision = 'ac8b1871be3d'
 down_revision = None
 branch_labels = None
 depends_on = None
@@ -18,10 +18,16 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('plate',
+    op.create_table('AllowedPlates',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('plate', sa.String(length=40), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('LoggedItems',
+    sa.Column('dateLogged', sa.DateTime(), nullable=False),
+    sa.Column('allowed', sa.Boolean(), server_default=sa.text('0'), nullable=False),
     sa.Column('id', sa.Integer(), nullable=False),
     sa.Column('plate', sa.String(length=40), nullable=False),
-    sa.Column('dateLogged', sa.DateTime(), nullable=True),
     sa.PrimaryKeyConstraint('id')
     )
     # ### end Alembic commands ###
@@ -29,5 +35,6 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table('plate')
+    op.drop_table('LoggedItems')
+    op.drop_table('AllowedPlates')
     # ### end Alembic commands ###

models.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+from application import db
+
+
+# db classes
+class Plate(db.Model):
+    __abstract__ = True
+    id = db.Column(db.Integer, primary_key=True)
+    plate = db.Column(db.String(40), nullable=False)
+
+
+class LoggedItem(Plate):
+    __tablename__ = "LoggedItems"
+    dateLogged = db.Column(db.DateTime, nullable=False)
+    allowed = db.Column(db.Boolean, nullable=False, server_default=db.false())
+
+    def __init__(
+        self,
+        plate: str,
+        datetime: datetime,
+        allowed: bool = False,
+    ):
+        self.plate = plate
+        self.allowed = allowed
+        self.dateLogged = datetime
+
+
+class AllowedPlate(Plate):
+    __tablename__ = "AllowedPlates"
+
+    def __init__(self, plate: str):
+        self.plate = plate

seed.py

@@ -0,0 +1,6 @@
+from application import db, app
+from models import AllowedPlate
+
+with app.app_context():
+    db.session.add(AllowedPlate("MUN389"))
+    db.session.commit()

server.py

@@ -1,86 +1,69 @@
-from flask import Flask, request, jsonify
+from flask import request, jsonify
-from flask_sqlalchemy import SQLAlchemy
+from pyplatex import ANPR  # type: ignore
-from flask_migrate import Migrate
+from ultralytics.nn.tasks import DetectionModel  # type: ignore
-from datetime import datetime
-from pyplatex import ANPR
-from ultralytics.nn.tasks import DetectionModel
 import os
 import torch
 import asyncio
+from application import app, db
+from models import LoggedItem, AllowedPlate
+from datetime import datetime
 
-torch.serialization.add_safe_globals({"DetectionModel": DetectionModel})
+torch.serialization.add_safe_globals(  # type: ignore
+    {"DetectionModel": DetectionModel}  # type: ignore
+)
 
-# 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!"
 
 
+i = 0
+
+
 # API to process vehicle
 @app.route("/api", methods=["POST"])
 def data():
-    # Check if image is present
+    data = request.data
-    if "image" not in request.files:
+    global i
-        return jsonify({"error": "No file part"}), 400
 
-    file = request.files["image"]  # get the image from the packet
+    with open(f"image{i}.jpg", "wb") as f:
+        f.write(data)
 
-    if file.filename == "" or not file.filename:
+    status = asyncio.run(process_image(f"image{i}.jpg"))
-        return jsonify({"error": "No selected file"}), 400
 
-    # Check if image ends in .jpg
+    i += 1
-    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",
+            "message": "Image sent succesfully",
-                "filename": file.filename,
+            "status": status,
-                "status": True,
-                "anpr": plate,
         }
     )
 
-    return jsonify({"error": "Only JPEG files allowed"}), 400
 
-
+async def process_image(file: str) -> bool:
-async def process_image(file: str):
     anpr = ANPR()
-    anpr_info = await anpr.detect(file)
+    anpr_info: ... = await anpr.detect(file)  # type: ignore
     number_plate = anpr_info["plate_number"]
     if number_plate:
-        db.session.add(LoggedItem(plate=number_plate))
+        allowed = (
+            AllowedPlate.query.filter_by(plate=number_plate).first()
+            is not None
+        )
+        db.session.add(
+            LoggedItem(
+                plate=number_plate, allowed=allowed, datetime=datetime.now()
+            )
+        )
         db.session.commit()
-        return number_plate
+        return allowed
-    return "ERROR"
+    return False
 
 
 if __name__ == "__main__":