Add application file

.gitignore

@@ -0,0 +1,2 @@
+.venv
+.env

Dockerfile

@@ -0,0 +1,14 @@
+# read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.9
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+COPY . .
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

config/database.py

@@ -0,0 +1,15 @@
+import pymongo
+import gridfs
+
+
+# Connect to MongoDB with the custom SSL context
+client = pymongo.MongoClient("mongodb+srv://soham:Soham2001@cluster0.pgjgiww.mongodb.net/?retryWrites=true&w=majority", ssl=True)
+
+db = client.database
+
+# Define collections
+admin_collection = db.admin_users
+user_collection = db.normal_users
+notification_collection = db.notifications
+pothole_image_collection = db.pothole_images
+fs = gridfs.GridFS(db)  # Initialize GridFS bucket

main.py

@@ -0,0 +1,14 @@
+from fastapi import FastAPI
+from routes.route import router
+from fastapi.middleware.cors import CORSMiddleware
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allow requests from any origin
+    allow_credentials=True,
+    allow_methods=["GET", "POST", "PUT", "DELETE"],
+    allow_headers=["Authorization", "Content-Type"],
+)
+
+app.include_router(router)

model/pothole_model.py

@@ -0,0 +1,67 @@
+# from ultralyticsplus import YOLO, render_result
+# from PIL import Image
+# import numpy as np
+
+# def load_model(image):
+#     # image_bytes = image.content
+#     model = YOLO('keremberke/yolov8n-pothole-segmentation')
+#     model.overrides['conf'] = 0.25
+#     model.overrides['iou'] = 0.45
+#     model.overrides['agnostic_nms'] = False
+#     model.overrides['max_det'] = 1000
+
+#     # Load image using PIL
+#     image = Image.open((image))
+#     image_array = np.array(image)
+#     # pil_image = pil_image.convert("RGB")  # Ensure image is in RGB format
+
+#     # Convert PIL image to bytes
+#     # with io.BytesIO() as output:
+#     #     pil_image.save(output, format='JPEG')
+#     #     image_bytes = output.getvalue()
+
+#     results = model.predict(image_array)
+#     for result in results:
+#         boxes = result.boxes.xyxy
+#         conf = result.boxes.conf
+#         cls = result.boxes.cls
+#         obj_info = []
+#         for i, bbox in enumerate(boxes):
+#             label = result.names[int(cls[i])]
+#             obj_info.append({
+#                 "Object": i+1,
+#                 "Label": label,
+#                 "Confidence": conf[i],
+#                 "Bounding Box": bbox
+#             })
+#     render = render_result(model=model, image=image, result=results[0])
+#     if label:
+#         print(label)
+#     render.show()
+#     return label
+
+
+from PIL import Image
+from io import BytesIO
+from transformers import AutoImageProcessor, AutoModelForImageClassification
+
+# Load model
+processor = AutoImageProcessor.from_pretrained("taroii/pothole-detection-model")
+model = AutoModelForImageClassification.from_pretrained("taroii/pothole-detection-model")
+
+# Function to predict if an image contains a pothole
+def predict_pothole(image_url):
+    image = Image.open(BytesIO(image_url))
+    inputs = processor(images=image, return_tensors="pt")
+
+    # Perform inference
+    outputs = model(**inputs)
+    logits = outputs.logits
+    probabilities = logits.softmax(dim=1)
+    
+    # Get predicted class (0: No pothole, 1: Pothole)
+    predicted_class = probabilities.argmax().item()
+    confidence = probabilities[0, predicted_class].item()
+
+    return predicted_class
+

pothole-detection-c31a0-firebase-adminsdk-xirxs-4ad5f554ca.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "pothole-detection-c31a0",
+  "private_key_id": "4ad5f554ca85b7542a5565af3b7edf63ee2367ed",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCSevQfEy+Rx5zn\nZ6XuYj+6/QjTroD2qL2T2p5ctytKGUqh4mSNHPcZqqKPjWesxo7ThNXqFFzKWFYe\nW4QCctIObe72T6zx/03jL2yhjGg9QhwefCJQ53CRmlnvg7amWEZL7ruNu2ede/ZQ\nsgngBhiqavN6yOYTVQKrxFO2j2NKHEHkBUSUDJtkOVzz2nXJtq4LCkKD8mgqup31\nB8ecEV1x32vGtaq/a1Y9HtMlVG+0Dmj1lL0pwHad8YPgjxaCzWvbhCWD2zF4R6FK\ns8mNIlRb8o42gAC+oLjX2gcktm0hDq+8t+z9nAbHdRFtYXHEFhan+Vw5xt7jvnLT\nqqndGybzAgMBAAECggEAJv+uTDG2+gibuS5qyqeE0Tt6JKWytzkg1CMiA4xO7bkD\nGkukur9J+J7qDe83eZQZxa35qAtI0ySmtQlngaFhVK8nLpPy9zEiYwim0vIHdLvL\nscAaANFFsrZWGINIV0xsVt4WODt7cD3nS//Cyk7FNWYpMfkX7HC3N9Ua9qGBbWLm\nbtsNMR39Wa9e5X5MWCSmIopiSJQq0AEVAaFbjMhcjwpbawket9joWkpz6zv3hSV+\nD9I7brQQVZNv504JfCNHdr4Ztth5Au5pMX3AQnGjGQ+SEv3XFE1QnMuAmmmPfZL9\n8bBygd5QpEJAlYYuOwZjDytpixvDoHf57nmiceALbQKBgQDHECqGax9RizwT7vN4\nRuWVxQ3ohuqoj8q7Z7L9coqsusWU6Jq74eefmW981oyoBPEniEMrM+3CuI2s3DQT\n2YnEtt8MQiwDh9v3lAy09ooZu2uHVhXF0cLW80d/peQu+OKRuDCsQIShOZOTBCLr\nZBXJjpjHC+w/QoN3mpfGf7pBFwKBgQC8YI6Hm5sgjSmdXnhw57jJd/jcupjpjnt2\n6OCpAIM3JYWRNw1X4tppx02tzMpWJSI2KW97bOYmSiTtgVhSOxj6h5rbwN71yNC4\nWZkQsof6vGxdxsb/pFiIOZBrBM8Ac/HgosHkmk6qPwszVsxHk7AiLRT0PpFktkSm\nx8jvtf0ahQKBgQC3ir1bASx4WGQFkR4mOWB4jp+7VaLJ1wM7dmr/63vXtcJ2AV6i\ne7HAY437UiuQxDXCmI4uKIxCLT5cMxFFeB/iJCdWuzCv9LWp+nUY6fT9suR7fbFH\nm5gF73xuQvw5Hzxw1Z7zQ+3GZjsepAK6fablGDjf1qt2zJJBVmY5HQ4T8wKBgQCW\nZfDlzBKadpnXDunSGu+pyqNid4hGUH+6fEuCIuqgSNDPE8kPdgszkPAv5uVUyej6\nnTJotJU71M6O4UoGvTBANawp718TlPUvejl+30s5oN5UMLQIvsoAWRU4nGo4zlk7\nJDbxsVFFE3h50L7gFsX1Q1ELgDjM08kT70Y6PG6LkQKBgQCGVZliwU+D90nWdY5N\ndSpVuE4V5jSUZ2W9jT1nX82Oy0leB4wkGosyUZOnMa8WNF24IJYPRP3U5GPsgAoH\npSwKGZAsuc7HWQ+q9bJ3b6hpWh0umaROhYOl5xm5KQLDII+mUk54SS2GGfWL8EEC\ncf5yQfuvn0pwmGymirQkqguS6g==\n-----END PRIVATE KEY-----\n",
+  "client_email": "firebase-adminsdk-xirxs@pothole-detection-c31a0.iam.gserviceaccount.com",
+  "client_id": "109748377203744803931",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-xirxs%40pothole-detection-c31a0.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

requirement.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn
+pydantic
+pymongo[srv]
+passlib[bcrypt]
+python-jose[cryptography]
+python-multipart
+certifi
+firebase-admin
+transformers
+pillow
+torch

routes/route.py

@@ -0,0 +1,273 @@
+from fastapi import APIRouter, HTTPException,Depends,File, UploadFile
+from fastapi.responses import JSONResponse
+from config.database import admin_collection, user_collection,notification_collection,pothole_image_collection
+from model.pothole_model import predict_pothole
+from utils.auth import create_access_token, hash_password, verify_password, verify_token
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from schema.model import Admin, PoholeInfo, PotInfoById, PotholeFilters, PotholeModel, UpdatePotholeInfo, User, UserLogin, VerifyOtp
+from utils.email_validator import send_activation_email
+import requests
+import uuid
+from bson import ObjectId
+import random
+security_scheme = HTTPBearer()
+router = APIRouter()
+activation_tokens={}
+import firebase_admin
+from firebase_admin import credentials, storage
+
+# Initialize Firebase Admin SDK
+cred = credentials.Certificate("pothole-detection-c31a0-firebase-adminsdk-xirxs-4ad5f554ca.json")
+firebase_admin.initialize_app(cred, {
+    'storageBucket': 'pothole-detection-c31a0.appspot.com'
+})
+
+def generate_otp():
+    return str(random.randint(100000, 999999))
+
+#----------------------------------------Admin Collections----------------------------------------
+# Admin Registration Api
+@router.post("/api/admin/registerAdmin", tags=["admin"])
+async def register_admin(admin: Admin):
+    hashed_password = hash_password(admin.password)
+    admin_dict = admin.dict()
+    admin_dict['password'] = hashed_password
+    activation_otp = generate_otp() 
+    admin_dict['otp'] = activation_otp
+    result = admin_collection.insert_one(admin_dict)
+    return {"message": "Admin created successfully", "user_id": str(result.inserted_id)}
+
+#Api to get all admins
+# @router.get("/api/admins/getAllAdmin", tags=["admin"])
+# async def get_all_admins():
+#     admins = admin_collection.find({})
+#     admin_list = []
+#     for admin in admins:
+#         admin["_id"] = str(admin["_id"]) 
+#         admin_list.append(admin)
+#     return admin_list
+
+
+
+
+
+#-------------------------------------------User Collections-------------------------------------------
+# User Login API
+@router.post("/api/user/userLogin", tags=["user"])
+async def user_login(userLogin: UserLogin):
+    # Check if the provided username and password are for admin
+    if userLogin.userName == "admin001" and userLogin.password == "Admin@123":
+        collection = admin_collection
+    else:
+        collection = user_collection
+
+    user = collection.find_one({"userName": userLogin.userName})
+    if not user or not verify_password(userLogin.password, user.get('password', '')):
+        raise HTTPException(status_code=401, detail="Incorrect username or password")
+    
+    # Check if the user is verified
+    if not user.get('isVerified'):
+        raise HTTPException(status_code=403, detail="Account is not verified")
+    
+    # Extract additional user data
+    user_data = {
+        "email": user.get("email"),
+        "username": user.get("userName"),
+        "id": str(user.get("_id")),
+        "role":user.get("role")
+    }
+    
+    token = create_access_token({"sub": userLogin.userName})
+    response_data = {"access_token": token, "token_type": "bearer", "user_data": user_data}
+    return response_data
+
+
+
+# User Registration Api
+@router.post("/api/user/registerUser", tags=["user"])
+async def create_user(user: User):
+    hashed_password = hash_password(user.password)
+    user_dict = user.dict()
+    user_dict['password'] = hashed_password
+    activation_otp = generate_otp() 
+    user_dict['otp'] = activation_otp
+    # Send activation email
+    send_activation_email(user.email, activation_otp)
+    # result = user_collection.insert_one(user_dict)
+    # return {"message": "User created successfully. Activation email sent.",}
+    result = user_collection.insert_one(user_dict)
+    return {"message": "User created successfully", "userId": str(result.inserted_id)}
+
+# Otp Verification Api
+@router.post("/api/user/verifyOtp", tags=["user"])
+async def verify_otp(verifyOtp:VerifyOtp):
+    # Check if the user exists with the given email
+    user = user_collection.find_one({"email": verifyOtp.email})
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    # Check if the provided OTP matches the stored OTP
+    if user.get('otp') != verifyOtp.otp:
+        raise HTTPException(status_code=400, detail="Invalid OTP")
+    # Update the user's isVerified field to True
+    user_collection.update_one({"email": verifyOtp.email}, {"$set": {"isVerified": True}})
+    return {"message": "OTP verified successfully and user is verified"}
+
+
+# Api to show the registered users list
+@router.get("/api/user/getAllUsers", tags=["user"], dependencies=[Depends(security_scheme)])
+async def get_all_user(token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+    users = user_collection.find({})
+    user_list = []
+    for user in users:
+        user["_id"] = str(user["_id"])  # Convert ObjectId to string
+        user_list.append(user)
+    return user_list
+
+
+
+
+#-------------------------------------------Pothole Collections-------------------------------------------
+
+# Api to submit the information about pothole
+@router.post("/api/information/submitInformation", tags=["pothole"], dependencies=[Depends(security_scheme)])
+async def upload_file(potholeInfo:PoholeInfo,token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+    pothole_info = potholeInfo.dict()
+    notification_collection.insert_one(pothole_info)
+
+    return {"message": "information submitted sucesssuccessfully"}
+
+
+# Api to update the information about pothole
+@router.put("/api/information/updateInformation", tags=["pothole"], dependencies=[Depends(security_scheme)])
+async def update_pothole_information(update_data: UpdatePotholeInfo, token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+
+    try:
+        # Update the pothole information in the collection
+        result = notification_collection.update_one({"_id": ObjectId(update_data.infoID)}, {"$set": {"status": update_data.status, "assignee": update_data.assignee}})
+        
+        if result.modified_count == 1:
+            return {"message": "Pothole information updated successfully"}
+        else:
+            return {"message": "No changes were made. Pothole information remains unchanged."}
+    except Exception as e:
+        return JSONResponse(content={"message": f"Error occurred: {str(e)}"}, status_code=500)
+
+
+# @router.get("/api/information/getAllFiles", tags=["pothole"], dependencies=[Depends(security_scheme)])
+# async def get_all_user(token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+#     verify_token(token.credentials)
+#     files = pothole_image_collection.find({})
+#     file_list = []
+#     for file in files:
+#         file["_id"] = str(file["_id"])  # Convert ObjectId to string
+#         file_list.append(file)
+#     return file_list
+
+
+# Function to upload file to Firebase Storage
+import os
+
+async def upload_file_to_firebase(file: UploadFile, token: HTTPAuthorizationCredentials):
+    verify_token(token.credentials)
+    try:
+        # Generate a unique ID for the file
+        fileID = uuid.uuid4().hex
+        # Extract the filename from the path provided by Flutter
+        filename = os.path.basename(file.filename)
+        # Get reference to Firebase Storage bucket
+        bucket = storage.bucket()
+        # Create a blob object with the filename
+        blob = bucket.blob(fileID + "-" + filename)
+        # Upload the file
+        blob.upload_from_file(file.file)
+
+        blob.make_public()
+        # Get the public URL of the uploaded file
+        url = blob.public_url
+        # Insert the document into the collection with Firebase URL
+        pothole_image_collection.insert_one({"fileID": fileID, "url": url, "filename": filename})
+        return JSONResponse(content={"message": "file uploaded successfully", "file_id": fileID})
+    except Exception as e:
+        return JSONResponse(content={"message": f"Error occurred: {str(e)}"}, status_code=500)
+
+@router.post("/api/information/fileUpload", tags=["pothole"])
+async def upload_file_api(file: UploadFile = File(...), token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    return await upload_file_to_firebase(file, token)
+
+
+
+# Api to get all information about pothole
+@router.post("/api/information/getAllPotholeInformation", tags=["pothole"], dependencies=[Depends(security_scheme)])
+async def get_all_info_with_filters(filters: PotholeFilters, token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+    # Construct a filter query based on provided parameters
+    filter_query = {}
+    if filters.userID == "66142a506d8f2a0f116fd613":
+        # If userID is the specific ID, do not filter by userID
+        pass
+    elif filters.userID:
+        filter_query["userId"] = filters.userID
+    # Always apply the status filter
+    if filters.status:
+        filter_query["status"] = filters.status
+    # Apply the filter query to find relevant pothole information
+    pothole_informations = notification_collection.find(filter_query)
+    info_list = []
+    for info in pothole_informations:
+        info["_id"] = str(info["_id"])
+        file_id = info.get('fileID')
+        if file_id:
+            image_data = pothole_image_collection.find_one({"fileID": file_id})
+            if image_data:
+                info['image'] = image_data.get('url')
+            else:
+                print("No image data found for file ID:", file_id)  # Debugging print
+        info_list.append(info)
+    return info_list
+
+
+
+# Api to get information about pothole by unique id
+@router.post("/api/information/getPotholeInformationById", tags=["pothole"], dependencies=[Depends(security_scheme)])
+async def get_data_by_id(potHoleInfoById:PotInfoById,token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+    try:
+        object_id = ObjectId(potHoleInfoById.infoID)
+        data = notification_collection.find_one({"_id": object_id})
+        if data:
+            data["_id"] = str(data["_id"])
+            image_data = pothole_image_collection.find_one({"fileID": data["fileID"]})
+            if image_data:
+                data["image"] = image_data["url"]  # Append image to the response
+            return data
+        else:
+            raise HTTPException(status_code=404, detail="Data not found")
+    except:
+        raise HTTPException(status_code=400, detail="Invalid ID format")
+    
+
+
+
+# Api to to verify the given object has pothole or not
+@router.post("/api/information/verifyPothole", tags=["pothole"], dependencies=[Depends(security_scheme)])
+async def verify_pothole(potholeModel: PotholeModel, token: HTTPAuthorizationCredentials = Depends(security_scheme)):
+    verify_token(token.credentials)
+
+    try:
+        # Get image bytes from URL
+        response = requests.get(potholeModel.image)
+        image_bytes = response.content
+
+        # Pass image bytes to your model function
+        results = predict_pothole(image_bytes)
+
+        if results == 1:
+            return JSONResponse(content={"response": "Pothole"})
+        else:
+            return JSONResponse(content={"response": "notPothole"})
+    except Exception as e:
+        return JSONResponse(content={"response": f"{e}"})
+

schema/model.py

@@ -0,0 +1,61 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class Admin(BaseModel):
+    name : str
+    userName : str
+    email :str
+    password :str
+    phoneNo :str
+    otp : int
+    isVerified : bool = True
+    role:str = 'admin'
+    
+
+class User(BaseModel):
+    name : str
+    userName : str
+    email :str
+    password :str
+    phoneNo :str
+    otp : int
+    isVerified : bool = False
+    role:str = 'user'
+
+
+class PoholeInfo(BaseModel):
+    userId : str
+    name: str
+    email: str
+    phoneNo :str
+    date: str
+    description: str
+    address: str
+    status: str
+    assignee: str
+    fileID: str
+    # isActive: bool
+    # version: int
+
+class UpdatePotholeInfo(BaseModel):
+    infoID: str
+    status: str
+    assignee: str
+
+class UserLogin(BaseModel):
+    userName : str
+    password : str
+
+class VerifyOtp(BaseModel):
+    email : str
+    otp:str
+
+class PotInfoById(BaseModel):
+    infoID : str
+
+class PotholeModel(BaseModel):
+    image : str
+
+class PotholeFilters(BaseModel):
+    userID: Optional[str]
+    status: Optional[str]

utils/auth.py

@@ -0,0 +1,40 @@
+from fastapi import HTTPException, status
+from fastapi.security import HTTPBearer
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+import bcrypt
+
+security = HTTPBearer()
+
+# Secret key for JWT token
+SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+# Function to generate JWT token
+def create_access_token(data: dict):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+# Function to hash a password
+def hash_password(password: str) -> str:
+    hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
+    return hashed_password
+
+# Function to verify a password
+def verify_password(plain_password: str, hashed_password: bytes) -> bool:
+    return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password)
+
+
+def verify_token(token: str):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return payload
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token"
+        )

utils/email_validator.py

@@ -0,0 +1,37 @@
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+import smtplib
+
+# Function to send activation email with HTML content
+def send_activation_email(email: str, activation_otp: int):
+    sender_email = "chandratresoham@gmail.com"  # Update with your email address
+    sender_password = "wbuc okcv hzzn iwyx"  # Update with your email password
+  # Update with your website URL
+    
+    # HTML content for the email body
+    email_body = f"""
+    <html>
+        <body>
+            <p>
+                Hello,<br><br>
+                You have successfully registered to the system.<br><br>
+                Please enter below otp to verify your accout<br><br>
+                <p style="font-size:17px; font-weight:bold">{activation_otp}</p><br><br>
+                Thank you!<br>
+            </p>
+        </body>
+    </html>
+    """
+
+    # Create MIMEText object with HTML content
+    message = MIMEMultipart("alternative")
+    message['From'] = sender_email
+    message['To'] = email
+    message['Subject'] = "Activate your account"
+    message.attach(MIMEText(email_body, 'html'))
+
+    # Connect to SMTP server and send email
+    with smtplib.SMTP('smtp.gmail.com', 587) as server:  # Update with your SMTP server details
+        server.starttls()
+        server.login(sender_email, sender_password)
+        server.sendmail(sender_email, email, message.as_string())