Initialize FastAPI application with inventory management and scanning…

… features

.gitignore

@@ -0,0 +1,6 @@
+# Virtualenv
+/venv
+
+# Pycache
+/__pycache__
+/routers/__pycache__

crud.py

@@ -0,0 +1,15 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from models import Inventory
+
+def get_inventory(db: Session):
+    return db.query(Inventory).all()
+
+def update_stock_level(db: Session, item_id: int, quantity: int):
+    item = db.query(Inventory).filter(Inventory.id == item_id).first()
+    if item:
+        item.stock_level += quantity
+        item.last_updated = func.now()
+        db.commit()
+        db.refresh(item)
+    return item

database.py

@@ -0,0 +1,17 @@
+# database.py
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./inventory.db"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

main.py

@@ -0,0 +1,23 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from database import engine, Base
+from routers import inventory, scanning
+
+Base.metadata.create_all(bind=engine)
+
+app = FastAPI()
+
+origins = [
+    "http://localhost:3000"
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(inventory.router, prefix="/api")
+app.include_router(scanning.router, prefix="/api")

mock_data.py

@@ -0,0 +1,35 @@
+import random
+import time
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from models import Inventory
+from database import SessionLocal
+
+route_warehouse_to_fps = [
+    {"lat": 40.712776, "lon": -74.005974},  # Start (Warehouse)
+    {"lat": 40.713000, "lon": -74.004000},
+    {"lat": 40.714000, "lon": -74.003000},
+    {"lat": 40.730610, "lon": -73.935242},  # End (FPS)
+]
+
+
+def mock_inventory_movement():
+    db = SessionLocal()
+    try:
+        while True:
+            item_id = random.randint(1, 10)  # Adjust range to match item IDs
+            quantity_change = random.choice([-1, 1]) * random.randint(1, 5)
+
+            item = db.query(Inventory).filter(Inventory.id == item_id).first()
+            if item:
+                item.stock_level += quantity_change
+                item.last_updated = func.now()
+                db.commit()
+                print(f"Updated stock for item {item.item_name}: {item.stock_level}")
+
+            time.sleep(5)  # Run every 5 seconds
+    finally:
+        db.close()
+
+if __name__ == "__main__":
+    mock_inventory_movement()

models.py

@@ -0,0 +1,37 @@
+from sqlalchemy import Column, Integer, String, DateTime, func, ForeignKey, Float
+from sqlalchemy.orm import relationship
+from database import Base
+
+class Inventory(Base):
+    __tablename__ = "inventory"
+
+    id = Column(Integer, primary_key=True, index=True)
+    item_name = Column(String, index=True)
+    stock_level = Column(Integer, default=0)
+    last_updated = Column(DateTime, default=func.now(), onupdate=func.now())
+
+
+class Journey(Base):
+    __tablename__ = "journey"
+    id = Column(Integer, primary_key=True, index=True)
+    item_id = Column(Integer, ForeignKey("inventory.id"))
+    checkpoint = Column(String, index=True)  # e.g., "Warehouse", "FPS"
+    timestamp = Column(DateTime, default=func.now())
+
+    item = relationship("Inventory")
+
+class GPSLocation(Base):
+    __tablename__ = "gps_location"
+    id = Column(Integer, primary_key=True, index=True)
+    vehicle_id = Column(String, index=True)
+    latitude = Column(Float)
+    longitude = Column(Float)
+    timestamp = Column(DateTime, default=func.now())
+
+class Alert(Base):
+    __tablename__ = 'alerts'
+    id = Column(Integer, primary_key=True, index=True)
+    vehicle_id = Column(String, index=True)
+    alert_type = Column(String)
+    description = Column(String)
+    timestamp = Column(DateTime, default=func.now())

routers/gps_tracking.py

@@ -0,0 +1,77 @@
+# gps_tracking.py
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from models import GPSLocation, Alert
+from database import get_db
+from mock_data import route_warehouse_to_fps
+from geopy.distance import geodesic
+from datetime import datetime
+
+MAX_DEVIATION_METERS = 500
+MAX_DELAY_SECONDS = 300
+
+router = APIRouter()
+
+route_index = 0
+
+@router.post("/gps/update")
+def update_gps_location(vehicle_id: str, db: Session = Depends(get_db)):
+    global route_index
+    if route_index >= len(route_warehouse_to_fps):
+        route_index = 0
+
+    # Get the next GPS location from the route
+    gps_data = route_warehouse_to_fps[route_index]
+    route_index += 1
+
+    # Create GPS location record
+    gps_location = GPSLocation(
+        vehicle_id=vehicle_id,
+        latitude=gps_data["lat"],
+        longitude=gps_data["lon"],
+        timestamp=func.now()
+    )
+    db.add(gps_location)
+    db.commit()
+
+    # Check for deviations and delays
+    if route_index > 1:
+        last_location = route_warehouse_to_fps[route_index - 2]
+        current_distance = geodesic(
+            (gps_data["lat"], gps_data["lon"]),
+            (last_location["lat"], last_location["lon"])
+        ).meters
+
+        # Alert for route deviation
+        if current_distance > MAX_DEVIATION_METERS:
+            alert = Alert(
+                vehicle_id=vehicle_id,
+                alert_type="Deviation",
+                description=f"Vehicle deviated {current_distance:.2f} meters from the route."
+            )
+            db.add(alert)
+
+        # Alert for delay
+        last_timestamp = gps_location.timestamp  # Timestamp when the location was updated
+        current_time = datetime.utcnow()
+        time_difference = (current_time - last_timestamp).total_seconds()
+
+        if time_difference > MAX_DELAY_SECONDS:
+            alert = Alert(
+                vehicle_id=vehicle_id,
+                alert_type="Delay",
+                description="Vehicle delayed at checkpoint."
+            )
+            db.add(alert)
+
+    db.commit()
+
+    return {"message": f"Vehicle {vehicle_id} location updated.", "location": gps_data}
+
+
+@router.get("/alerts")
+def get_alerts(vehicle_id: str, db: Session = Depends(get_db)):
+    alerts = db.query(Alert).filter(Alert.vehicle_id == vehicle_id).all()
+    return {"vehicle_id": vehicle_id, "alerts": alerts}

routers/inventory.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from pydantic import BaseModel
+import crud
+from database import get_db
+
+router = APIRouter()
+
+class UpdateStockRequest(BaseModel):
+    item_id: int
+    quantity: int
+
+@router.get("/inventory")
+def get_inventory(db: Session = Depends(get_db)):
+    return crud.get_inventory(db)
+
+@router.post("/inventory/update")
+def update_stock(data: UpdateStockRequest, db: Session = Depends(get_db)):
+    item = crud.update_stock_level(db, data.item_id, data.quantity)
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"message": "Stock level updated", "new_stock_level": item.stock_level}

routers/scanning.py

@@ -0,0 +1,28 @@
+# routers/scanning.py
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from models import Journey, Inventory
+from database import get_db
+import random
+import datetime
+
+router = APIRouter()
+
+checkpoints = ["Warehouse", "In Transit", "FPS"]
+
+@router.post("/scan")
+def simulate_scan(item_id: int, db: Session = Depends(get_db)):
+    # Get the item
+    item = db.query(Inventory).filter(Inventory.id == item_id).first()
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+
+    # Generate a random checkpoint
+    checkpoint = random.choice(checkpoints)
+
+    # Create a new journey record
+    journey_entry = Journey(item_id=item.id, checkpoint=checkpoint, timestamp=datetime.datetime.utcnow())
+    db.add(journey_entry)
+    db.commit()
+
+    return {"message": f"Item {item.item_name} scanned at {checkpoint} checkpoint"}

seed.py

@@ -0,0 +1,24 @@
+from database import SessionLocal, engine, Base
+from models import Inventory
+
+Base.metadata.create_all(bind=engine)
+
+def seed_inventory():
+    db = SessionLocal()
+    try:
+        if db.query(Inventory).count() == 0:
+            sample_items = [
+                Inventory(item_name="Rice", stock_level=100),
+                Inventory(item_name="Wheat", stock_level=200),
+                Inventory(item_name="Sugar", stock_level=150),
+                Inventory(item_name="Oil", stock_level=80)
+            ]
+            db.add_all(sample_items)
+            db.commit()
+            print("Sample inventory data added.")
+        else:
+            print("Inventory already seeded.")
+    finally:
+        db.close()
+
+seed_inventory()