Do0rMaMu's picture
Update app.py
dda1cd6 verified
raw
history blame
26.5 kB
import pandas as pd
from itertools import permutations, combinations
from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
import uuid
import requests
import json
from fastapi.responses import RedirectResponse
import pyodbc
import os
from dotenv import load_dotenv
from datetime import datetime
# Load environment variables from .env file
load_dotenv()
# Fetch credentials from environment variables
DB_DRIVER = os.getenv('DB_DRIVER')
DB_SERVER = os.getenv('DB_SERVER')
DB_DATABASE = os.getenv('DB_DATABASE')
DB_UID = os.getenv('DB_UID')
DB_PWD = os.getenv('DB_PWD')
DB_ENCRYPT = os.getenv('DB_ENCRYPT')
DB_NAME = os.getenv('DB_name')
DB_TABLE_NAME = os.getenv('DB_tableName')
frontend_url = os.getenv('FRONTEND_URL')
# Define connection string
conn_str = (
f"Driver={DB_DRIVER};"
f"Server={DB_SERVER};"
f"Database={DB_NAME};" # Use the target database
f"UID={DB_UID};"
f"PWD={DB_PWD};"
f"Encrypt={DB_ENCRYPT};"
)
app = FastAPI()
stored_combinations = {}
destination_mapping_global = {}
processed_data_store: Dict[str, Dict[str, Any]] = {} # Store processed data with session_id
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust as needed for security
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.options("/{rest_of_path:path}")
async def preflight_handler(request: Request, rest_of_path: str) -> Response:
return Response(headers={
"Access-Control-Allow-Origin": "*", # Adjust as needed for security
"Access-Control-Allow-Methods": "POST, GET, OPTIONS, PUT, DELETE",
"Access-Control-Allow-Headers": "*",
})
# Box and Truck classes
class Box:
def __init__(self, length, width, height, quantity, box_type, destination):
self.length = length
self.width = width
self.height = height
self.quantity = quantity if quantity > 0 else 1
self.type = box_type
self.destination = destination # Added Destination
def rotated_dimensions(self):
possible_rotations = [
(self.length, self.width, self.height),
(self.length, self.height, self.width),
(self.width, self.length, self.height),
(self.width, self.height, self.length),
(self.height, self.length, self.width),
(self.height, self.width, self.length)
]
return sorted(possible_rotations, key=lambda x: (x[0] * x[1], x[2]), reverse=True)
def volume(self):
return self.length * self.width * self.height
class Truck:
def __init__(self, length, width, height):
self.length = length
self.width = width
self.height = height
self.volume = length * width * height
self.placed_boxes = []
self.available_space = [(0, 0, 0, length, width, height)]
def is_space_supported(self, position, box):
x, y, z = position
bl, bw, bh = box
if z == 0:
return True
for placed_box, placed_position in self.placed_boxes:
px, py, pz = placed_position
pl, pw, ph = placed_box
if (x >= px and x + bl <= px + pl) and (y >= py and y + bw <= py + pw) and (z == pz + ph):
return True
return False
def add_box_ffd(self, box):
best_fit = None
best_fit_space_index = -1
best_fit_volume_waste = float('inf')
for rotation in box.rotated_dimensions():
for i, space in enumerate(self.available_space):
x, y, z, l, w, h = space
if rotation[0] <= l and rotation[1] <= w and rotation[2] <= h:
if not self.is_space_supported((x, y, z), rotation):
continue
leftover_volume = (l * w * h) - (rotation[0] * rotation[1] * rotation[2])
if leftover_volume < best_fit_volume_waste:
best_fit = rotation
best_fit_space_index = i
best_fit_volume_waste = leftover_volume
if best_fit:
x, y, z, l, w, h = self.available_space[best_fit_space_index]
box_position = (x, y, z)
self.placed_boxes.append((best_fit, box_position))
self.available_space.pop(best_fit_space_index)
self.update_space(best_fit, box_position, l, w, h)
return box_position
else:
return None
def update_space(self, box, position, l, w, h):
x, y, z = position
bl, bw, bh = box
new_spaces = [
(x + bl, y, z, l - bl, w, h),
(x, y + bw, z, bl, w - bw, h),
(x, y, z + bh, bl, bw, h - bh),
]
new_spaces = [space for space in new_spaces if space[3] > 0 and space[4] > 0 and space[5] > 0]
self.available_space.extend(new_spaces)
self.available_space.sort(key=lambda space: (space[2], space[1], space[0]))
def pack_boxes_ffd(truck, consignments):
packed_positions = []
# Sort consignments: priority consignments first, then by total volume (descending)
consignments = sorted(consignments, key=lambda c: (not c[1], -sum(box.volume() * box.quantity for box in c[0])))
for consignment in consignments:
consignment_boxes = consignment[0]
for box in consignment_boxes:
for _ in range(box.quantity):
position = truck.add_box_ffd(box)
if position is not None:
# Box is packed
packed_positions.append((box, position))
else:
# Box cannot be packed, include with position as None
packed_positions.append((box, None))
return packed_positions
def find_max_consignments_combinations(truck, consignments, combination_limit=100):
combinations_that_fit = []
consignments_sorted = sorted(consignments, key=lambda c: (not c[1], -sum(box.volume() * box.quantity for box in c[0])))
combo_count = 0
max_consignments_fitted = 0
for r in range(1, len(consignments_sorted) + 1):
for combo in combinations(consignments_sorted, r):
combo_count += 1
# Corrected line: Access the first box's type
if len(combo) == 0 or any(len(consignment[0]) == 0 for consignment in combo):
# Skip if any consignment has no boxes
continue
print(f"Checking combination {combo_count}: {[f'Consignment {boxes[0].type}' for boxes, _ in combo]}")
if combo_count > combination_limit:
print(f"Reached combination limit of {combination_limit}. Stopping further checks.")
break
temp_truck = Truck(truck.length, truck.width, truck.height)
total_volume = sum(sum(box.volume() * box.quantity for box in consignment[0]) for consignment in combo)
if total_volume > temp_truck.volume:
continue
try:
packed_positions = pack_boxes_ffd(temp_truck, combo)
if packed_positions:
print(f"Combination {combo_count} fits with {len(combo)} consignments")
combinations_that_fit.append(combo)
max_consignments_fitted = max(max_consignments_fitted, len(combo))
if len(combinations_that_fit) >= 5:
break
except HTTPException:
print(f"Combination {combo_count} does not fit")
continue
if combo_count > combination_limit:
break
return combinations_that_fit[-5:]
def suggest_truck(consignments):
trucks = {
"TATA ACE": {"length": 7, "width": 4.8, "height": 4.8},
"ASHOK LEYLAND DOST": {"length": 7, "width": 4.8, "height": 4.8},
"MAHINDRA BOLERO PICK UP": {"length": 8, "width": 5, "height": 4.8},
"ASHOK LEYLAND BADA DOST": {"length": 9.5, "width": 5.5, "height": 5},
"TATA 407": {"length": 9, "width": 5.5, "height": 5},
"EICHER 14 FEET": {"length": 14, "width": 6, "height": 6.5},
"EICHER 17 FEET": {"length": 17, "width": 6, "height": 7},
"EICHER 19 FEET": {"length": 19, "width": 7, "height": 7},
"TATA 22 FEET": {"length": 22, "width": 7.5, "height": 7},
"TATA TRUCK (6 TYRE)": {"length": 17.5, "width": 7, "height": 7},
"TAURUS 16 T (10 TYRE)": {"length": 21, "width": 7.2, "height": 7},
"TAURUS 21 T (12 TYRE)": {"length": 24, "width": 7.3, "height": 7},
"TAURUS 25 T (14 TYRE)": {"length": 28, "width": 7.8, "height": 7},
"CONTAINER 20 FT": {"length": 20, "width": 8, "height": 8},
"CONTAINER 32 FT SXL": {"length": 32, "width": 8, "height": 8},
"CONTAINER 32 FT MXL": {"length": 32, "width": 8, "height": 8},
"CONTAINER 32 FT SXL / MXL HQ": {"length": 32, "width": 8, "height": 10},
"20 FEET OPEN ALL SIDE (ODC)": {"length": 20, "width": 8, "height": 8},
"28-32 FEET OPEN-TRAILOR JCB ODC": {"length": 28, "width": 8, "height": 8},
"32 FEET OPEN-TRAILOR ODC": {"length": 32, "width": 8, "height": 8},
"40 FEET OPEN-TRAILOR ODC": {"length": 40, "width": 8, "height": 8},
"SCV": {"length": 5, "width": 12, "height": 6},
"LCV": {"length": 11, "width": 5, "height": 5},
"ICV": {"length": 16, "width": 6.5, "height": 6.5},
"MCV": {"length": 19, "width": 7, "height": 7}
}
sorted_trucks = sorted(trucks.items(), key=lambda t: t[1]['length'] * t[1]['width'] * t[1]['height'])
for truck_name, dimensions in sorted_trucks:
truck = Truck(dimensions['length'] * 12, dimensions['width'] * 12, dimensions['height'] * 12)
packed_positions = pack_boxes_ffd(truck, consignments)
total_boxes = sum(box.quantity for consignment in consignments for box in consignment[0])
# Count successfully packed boxes
packed_boxes_count = sum(1 for _, pos in packed_positions if pos is not None)
if packed_boxes_count == total_boxes:
return {"name": truck_name, "dimensions": dimensions}
return None
# Updated Pydantic Models with Destination Fields
class BoxData(BaseModel):
PieceLength: float
PieceBreadth: float
PieceHeight: float
Priority: int
Destination: str # Added Destination
class ConsignmentData(BaseModel):
ConsignmentNo: str
Priority: int
Destination: str # Added Destination
boxes: List[BoxData]
class SubmitDataRequest(BaseModel):
truck_name: str
autoSuggest: bool
consignments_data: List[ConsignmentData]
destination_mapping: Dict[str, int] # Added destination_mapping
@app.post("/submit_data/")
async def submit_data(request: SubmitDataRequest):
"""
Endpoint to receive data from the client via API.
"""
frontend_url = os.getenv('FRONTEND_URL')
truck_name = request.truck_name
autoSuggest = request.autoSuggest
consignments_data = request.consignments_data # This is already parsed from JSON
destination_mapping = request.destination_mapping # Received destination_mapping
global stored_combinations, processed_data_store, destination_mapping_global
destination_mapping_global = destination_mapping
# Parse consignments and calculate total quantity based on box dimensions
consignments = []
for consignment_data in consignments_data:
consignment_no = consignment_data.ConsignmentNo # ConsignmentNo is now accessible as a string
is_priority = consignment_data.Priority
consignment_destination = consignment_data.Destination # Get Destination for Consignment
# Aggregate the boxes with same dimensions and destination
box_aggregation = {}
for box_data in consignment_data.boxes:
dimensions = (box_data.PieceLength, box_data.PieceBreadth, box_data.PieceHeight, box_data.Destination)
if dimensions in box_aggregation:
box_aggregation[dimensions]['Quantity'] += 1 # Increment the count if dimensions are the same
else:
box_aggregation[dimensions] = {
'PieceLength': box_data.PieceLength,
'PieceBreadth': box_data.PieceBreadth,
'PieceHeight': box_data.PieceHeight,
'Destination': box_data.Destination, # Include Destination
'Quantity': 1 # Initialize count to 1
}
# Convert aggregated boxes back to the expected list format
consignment_boxes = [
Box(
box['PieceLength'] / 2.54, # Convert cm to inches
box['PieceBreadth'] / 2.54,
box['PieceHeight'] / 2.54,
box['Quantity'],
f"Consignment {consignment_no} ({'Priority' if is_priority else 'Non-Priority'})",
box['Destination'] # Pass Destination to Box
)
for box in box_aggregation.values()
]
consignments.append((consignment_boxes, is_priority))
# Handle autoSuggest logic and other functionality here
if autoSuggest:
suggested_truck = suggest_truck(consignments)
if suggested_truck:
truck_name = suggested_truck["name"]
length = suggested_truck['dimensions']['length']
width = suggested_truck['dimensions']['width']
height = suggested_truck['dimensions']['height']
else:
raise HTTPException(status_code=400, detail="No suitable truck found")
else:
# Assuming truck dimensions are known based on truck_name
trucks = {
"TATA ACE": {"length": 7, "width": 4.8, "height": 4.8},
"ASHOK LEYLAND DOST": {"length": 7, "width": 4.8, "height": 4.8},
"MAHINDRA BOLERO PICK UP": {"length": 8, "width": 5, "height": 4.8},
"ASHOK LEYLAND BADA DOST": {"length": 9.5, "width": 5.5, "height": 5},
"TATA 407": {"length": 9, "width": 5.5, "height": 5},
"EICHER 14 FEET": {"length": 14, "width": 6, "height": 6.5},
"EICHER 17 FEET": {"length": 17, "width": 6, "height": 7},
"EICHER 19 FEET": {"length": 19, "width": 7, "height": 7},
"TATA 22 FEET": {"length": 22, "width": 7.5, "height": 7},
"TATA TRUCK (6 TYRE)": {"length": 17.5, "width": 7, "height": 7},
"TAURUS 16 T (10 TYRE)": {"length": 21, "width": 7.2, "height": 7},
"TAURUS 21 T (12 TYRE)": {"length": 24, "width": 7.3, "height": 7},
"TAURUS 25 T (14 TYRE)": {"length": 28, "width": 7.8, "height": 7},
"CONTAINER 20 FT": {"length": 20, "width": 8, "height": 8},
"CONTAINER 32 FT SXL": {"length": 32, "width": 8, "height": 8},
"CONTAINER 32 FT MXL": {"length": 32, "width": 8, "height": 8},
"CONTAINER 32 FT SXL / MXL HQ": {"length": 32, "width": 8, "height": 10},
"20 FEET OPEN ALL SIDE (ODC)": {"length": 20, "width": 8, "height": 8},
"28-32 FEET OPEN-TRAILOR JCB ODC": {"length": 28, "width": 8, "height": 8},
"32 FEET OPEN-TRAILOR ODC": {"length": 32, "width": 8, "height": 8},
"40 FEET OPEN-TRAILOR ODC": {"length": 40, "width": 8, "height": 8},
"SCV": {"length": 5, "width": 12, "height": 6},
"LCV": {"length": 11, "width": 5, "height": 5},
"ICV": {"length": 16, "width": 6.5, "height": 6.5},
"MCV": {"length": 19, "width": 7, "height": 7}
}
if truck_name not in trucks:
raise HTTPException(status_code=400, detail="Invalid truck name provided")
truck_dimensions = trucks[truck_name]
length = truck_dimensions['length']
width = truck_dimensions['width']
height = truck_dimensions['height']
# Initialize the Truck object and pack boxes
truck = Truck(length * 12, width * 12, height * 12) # Convert feet to inches
packed_positions = pack_boxes_ffd(truck, consignments)
# Count how many boxes were successfully packed
packed_boxes_count = sum(1 for _, pos in packed_positions if pos is not None)
# Total number of boxes
total_boxes = len(packed_positions) # Since we expanded quantities during packing
if packed_boxes_count < total_boxes:
stored_combinations = find_max_consignments_combinations(truck, consignments)
if stored_combinations:
combinations_info = []
for index, combo in enumerate(stored_combinations):
total_quantity = sum(box.quantity for boxes, _ in combo for box in boxes)
combination_info = {
"combination_number": index + 1,
"consignments": [
{
"consignment_number": f"{boxes[0].type.split(' ')[1]}", # Corrected access
"priority_status": boxes[0].type.split('(')[1].strip(')'),
"destination": boxes[0].destination # Include Destination
}
for boxes, _ in combo
],
"total_quantity": total_quantity # Add total quantity for this combination
}
combinations_info.append(combination_info)
# Generate a unique session_id
session_id = str(uuid.uuid4())
processed_data_store[session_id] = {
"error": "Not all consignments fit.",
"combinations_that_fit": combinations_info,
"truck": {
"name": truck_name,
"dimensions": {
"length": length,
"width": width,
"height": height
}
},
"consignments_data": [consignment.dict() for consignment in consignments_data], # **Includes Destination**
"destination_mapping": destination_mapping # Include destination_mapping
}
# Redirect to the visualization page
return RedirectResponse(url=f"{frontend_url}?session_id={session_id}", status_code=302)
# Prepare box data for visualization
box_data = []
for box, pos in packed_positions:
if pos is not None:
position = {'x': pos[0], 'y': pos[1], 'z': pos[2]}
else:
position = None # Box could not be packed
box_data.append({
'length': box.length,
'width': box.width,
'height': box.height,
'type': box.type,
'quantity': 1, # Since we're iterating over each individual box
'position': position,
'destination': box.destination # Include Destination
})
# Generate a unique session_id
session_id = str(uuid.uuid4())
processed_data_store[session_id] = {
"boxes": box_data,
"truck": {
"name": truck_name,
"dimensions": {
"length": length,
"width": width,
"height": height
}
},
"destination_mapping": destination_mapping # Include destination_mapping
}
print(" auto suggest data ", processed_data_store[session_id])
# Redirect to the visualization page
return RedirectResponse(url=f"{frontend_url}?session_id={session_id}", status_code=302)
@app.get("/get_visualization/{session_id}/")
async def get_visualization(session_id: str):
"""
Endpoint for the frontend to fetch processed data using session_id.
"""
if session_id not in processed_data_store:
raise HTTPException(status_code=404, detail="Session ID not found")
return processed_data_store[session_id]
class SaveCombinationRequest(BaseModel):
combination_number: int
truck: str
truck_length: float
truck_width: float
truck_height: float
consignments: List[ConsignmentData]
total_quantity: int
utilization_percentage: Optional[float]
# Endpoint to save combination data
@app.post("/save_combination/", summary="Save a truck loading combination")
async def save_combination(data: SaveCombinationRequest):
try:
global destination_mapping_global
# Establish a connection
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()
# Prepare data for insertion
current_date = datetime.now().strftime("%d-%m-%Y")
truck_name = data.truck
# Truck dimensions are now captured separately
truck_length = data.truck_length
truck_width = data.truck_width
truck_height = data.truck_height
total_consignments = len(data.consignments)
total_boxes = sum(len(consignment.boxes) for consignment in data.consignments)
# Prepare box data as payload for storage
box_data = []
for consignment in data.consignments:
for box in consignment.boxes:
box_entry = {
'length': box.PieceLength,
'width': box.PieceBreadth,
'height': box.PieceHeight,
'type': consignment.ConsignmentNo,
'quantity': 1, # Assuming the quantity is handled individually in frontend
'position': "None",
'destination': box.Destination
}
box_data.append(box_entry)
payload = {
"boxes": box_data,
"truck": {
"name": truck_name,
"dimensions": {
"length": truck_length,
"width": truck_width,
"height": truck_height
}
},
"destination_mapping": destination_mapping_global,
"total_boxes": total_boxes,
"total_consignments": total_consignments
}
# print("payload ", payload)
# Insert data into the table
insert_query = f"""
INSERT INTO {DB_TABLE_NAME} (Date, truckName, truckDimensions, totalConsignment, totalBoxes, payload)
VALUES (?, ?, ?, ?, ?, ?)
"""
cursor.execute(insert_query, (
current_date,
truck_name,
f"{truck_length}L x {truck_width}W x {truck_height}H Feet",
total_consignments,
total_boxes, # Total number of boxes calculated
json.dumps(payload) # Store the payload as a JSON string
))
# Commit the transaction
conn.commit()
# Close the connection
cursor.close()
conn.close()
return {"message": "Combination saved successfully.", "Date": current_date}
except pyodbc.Error as e:
error_message = str(e)
raise HTTPException(status_code=500, detail=f"Database error: {error_message}")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/visualize_payload/")
async def visualize_payload(payload: Dict[str, Any]):
"""
API to receive a payload and return the processed data to the frontend via session ID.
"""
global processed_data_store
frontend_url = os.getenv('FRONTEND_URL')
destination_mapping = payload.get('destination_mapping')
# Extract truck details from payload
truck = payload.get('truck')
if not truck or 'dimensions' not in truck:
raise HTTPException(status_code=400, detail="Truck information or dimensions missing.")
truck_name = truck.get('name')
truck_dimensions = truck.get('dimensions')
if not truck_dimensions:
raise HTTPException(status_code=400, detail="Truck dimensions missing.")
# Extract boxes from payload
boxes = payload.get('boxes', [])
if not boxes:
raise HTTPException(status_code=400, detail="No boxes found in payload.")
# Extract and process truck dimensions
length = truck_dimensions.get('length')
width = truck_dimensions.get('width')
height = truck_dimensions.get('height')
if not length or not width or not height:
raise HTTPException(status_code=400, detail="Incomplete truck dimensions.")
# Initialize Truck object
truck_obj = Truck(length * 12, width * 12, height * 12) # Convert feet to inches
# Process the boxes for FFD packing
packed_positions = []
consignments = []
for box in boxes:
# Convert box dimensions to inches (assuming they are given in feet or inches depending on your context)
box_obj = Box(
length=box['length'],
width=box['width'],
height=box['height'],
quantity=box.get('quantity', 1),
box_type=box['type'],
destination=box['destination']
)
# Use pack_boxes_ffd method to find positions for boxes in truck
position = truck_obj.add_box_ffd(box_obj)
packed_positions.append((box_obj, position))
# Prepare the box data for visualization
box_data = []
for box_obj, pos in packed_positions:
if pos is not None:
position = {'x': pos[0], 'y': pos[1], 'z': pos[2]}
else:
position = None # Box could not be packed
box_data.append({
'length': box_obj.length / 2.54,
'width': box_obj.width / 2.54,
'height': box_obj.height / 2.54,
'type': box_obj.type,
'quantity': box_obj.quantity,
'position': position,
'destination': box_obj.destination
})
# Generate a unique session_id for storing the processed data
session_id = str(uuid.uuid4())
processed_data_store[session_id] = {
"boxes": box_data,
"truck": {
"name": truck_name,
"dimensions": {
"length": length,
"width": width,
"height": height
}
},
"destination_mapping": destination_mapping
}
# Return session ID and processed data to the frontend for visualization
return {"session_id": session_id, "visualization_url": f"{frontend_url}?session_id={session_id}"}