Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import requests | |
| import os | |
| from PIL import Image | |
| import numpy as np | |
| from transformers import pipeline | |
| from simple_salesforce import Salesforce | |
| import io | |
| import time | |
| from dotenv import load_dotenv | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # Function to validate photo size (< 20MB) | |
| def validate_photo_size(image_file): | |
| max_size_mb = 20 | |
| if isinstance(image_file, Image.Image): | |
| # Convert PIL Image to bytes for size check | |
| img_byte_arr = io.BytesIO() | |
| image_file.save(img_byte_arr, format='JPEG') | |
| file_size_mb = img_byte_arr.tell() / (1024 * 1024) # Convert bytes to MB | |
| return file_size_mb <= max_size_mb, None | |
| return False, "Invalid image format" | |
| # Function to process image with AI and predict milestone | |
| def predict_milestone(image): | |
| try: | |
| # Simulate AI processing time (ensure < 5 seconds) | |
| start_time = time.time() | |
| # Process image with Hugging Face model | |
| model = pipeline("image-classification", model="microsoft/resnet-50") | |
| predictions = model(image) | |
| # Placeholder logic: Map model output to construction milestones | |
| milestone = predictions[0]["label"] # Example: "positive" -> "Walls Erected" | |
| confidence = predictions[0]["score"] | |
| # Map model output to construction milestones (customize this) | |
| milestone_map = { | |
| "positive": "Walls Erected", | |
| "negative": "Foundation Completed", | |
| # Add more mappings based on your model | |
| } | |
| completion_map = { | |
| "positive": 60.00, # Example: Walls = 60% complete | |
| "negative": 20.00, # Example: Foundation = 20% complete | |
| } | |
| predicted_milestone = milestone_map.get(milestone, "Unknown Milestone") | |
| completion_percentage = completion_map.get(milestone, 0.00) | |
| processing_time = time.time() - start_time | |
| if processing_time > 5: | |
| return None, None, "AI took too long to process (> 5 seconds)." | |
| return predicted_milestone, completion_percentage, None | |
| except Exception as e: | |
| return None, None, f"AI failed to process the image: {str(e)}" | |
| # Function to upload image to Salesforce and get a URL | |
| def upload_image_to_salesforce(image, project_name): | |
| try: | |
| # Placeholder: Simulate uploading image to Salesforce ContentVersion | |
| image_url = f"https://your-salesforce-instance.com/file/{project_name}.jpg" # Simulated URL | |
| return image_url, None | |
| except Exception as e: | |
| return None, f"Failed to upload image to Salesforce: {str(e)}" | |
| # Function to update Salesforce Construction_Project__c object and fetch fields | |
| def update_salesforce_record(sf, project_name, milestone, percentage, image_url, status, comments): | |
| try: | |
| # Query to check if the project exists | |
| query = f"SELECT Id FROM Construction_Project__c WHERE Name = '{project_name}'" | |
| result = sf.query(query) | |
| if result['totalSize'] == 0: | |
| return None, f"No project found with Name: {project_name}" | |
| record_id = result['records'][0]['Id'] | |
| # Update the record | |
| sf.Construction_Project__c.update(record_id, { | |
| 'Current_Milestone__c': milestone, | |
| 'Completion_Percentage__c': percentage, | |
| 'Last_Updated_Image__c': image_url, | |
| 'Last_Updated_On__c': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), | |
| 'Upload_Status__c': status, | |
| 'Comments__c': comments | |
| }) | |
| # Fetch the updated record to get the specified fields | |
| updated_query = f"SELECT Current_Milestone__c, Last_Updated_Image__c, Last_Updated_On__c, Upload_Status__c FROM Construction_Project__c WHERE Id = '{record_id}'" | |
| updated_result = sf.query(updated_query) | |
| if updated_result['totalSize'] == 0: | |
| return None, "Failed to retrieve updated record." | |
| record = updated_result['records'][0] | |
| fields_output = { | |
| 'Current_Milestone__c': record.get('Current_Milestone__c', 'N/A'), | |
| 'Last_Updated_Image__c': record.get('Last_Updated_Image__c', 'N/A'), | |
| 'Last_Updated_On__c': record.get('Last_Updated_On__c', 'N/A'), | |
| 'Upload_Status__c': record.get('Upload_Status__c', 'N/A') | |
| } | |
| return fields_output, None | |
| except Exception as e: | |
| return None, f"Failed to update Salesforce: {str(e)}" | |
| # Main Gradio function | |
| def process_construction_photo(project_name, image): | |
| if not project_name or not image: | |
| return None, "Please provide a project name and upload a photo." | |
| # Connect to Salesforce | |
| try: | |
| sf = Salesforce( | |
| username=os.getenv('SALESFORCE_USERNAME'), | |
| password=os.getenv('SALESFORCE_PASSWORD'), | |
| security_token=os.getenv('SALESFORCE_SECURITY_TOKEN'), | |
| domain=os.getenv('SALESFORCE_DOMAIN') | |
| ) | |
| except Exception as e: | |
| return None, f"Failed to connect to Salesforce: {str(e)}" | |
| # Validate photo size | |
| is_valid, error = validate_photo_size(image) | |
| if not is_valid: | |
| return None, error or "Photo is too large! Please upload a photo smaller than 20MB." | |
| # Process the image with AI | |
| milestone, percentage, error = predict_milestone(image) | |
| if error: | |
| fields, error_message = update_salesforce_record( | |
| sf=sf, | |
| project_name=project_name, | |
| milestone=None, | |
| percentage=0.00, | |
| image_url=None, | |
| status="Failure", | |
| comments=error | |
| ) | |
| error_text = f"AI Error: {error}" | |
| if error_message: | |
| error_text += f"\nSalesforce Error: {error_message}" | |
| if fields: | |
| error_text += "\nUpdated Salesforce Fields:\n" | |
| for field, value in fields.items(): | |
| error_text += f"{field}: {value}\n" | |
| return None, error_text | |
| # Upload image to Salesforce | |
| image_url, upload_error = upload_image_to_salesforce(image, project_name) | |
| if upload_error: | |
| fields, error_message = update_salesforce_record( | |
| sf=sf, | |
| project_name=project_name, | |
| milestone=milestone, | |
| percentage=percentage, | |
| image_url=None, | |
| status="Failure", | |
| comments=upload_error | |
| ) | |
| error_text = f"Upload Error: {upload_error}" | |
| if error_message: | |
| error_text += f"\nSalesforce Error: {error_message}" | |
| if fields: | |
| error_text += "\nUpdated Salesforce Fields:\n" | |
| for field, value in fields.items(): | |
| error_text += f"{field}: {value}\n" | |
| return None, error_text | |
| # Update Salesforce with success | |
| fields, error_message = update_salesforce_record( | |
| sf=sf, | |
| project_name=project_name, | |
| milestone=milestone, | |
| percentage=percentage, | |
| image_url=image_url, | |
| status="Success", | |
| comments="Photo processed successfully" | |
| ) | |
| if error_message: | |
| return None, f"Salesforce Error: {error_message}" | |
| # Prepare output with AI results and Salesforce fields | |
| result_text = f"Success! Milestone: {milestone}, Completion: {percentage}%\nProgress saved to Salesforce!\n\nSalesforce Fields:\n" | |
| for field, value in fields.items(): | |
| result_text += f"{field}: {value}\n" | |
| return image, result_text | |
| # Gradio interface | |
| iface = gr.Interface( | |
| fn=process_construction_photo, | |
| inputs=[ | |
| gr.Textbox(label="Project Name (e.g., Sunshine Apartments)", placeholder="Sunshine Apartments"), | |
| gr.Image(type="pil", label="Upload a Construction Photo") | |
| ], | |
| outputs=[ | |
| gr.Image(label="Uploaded Photo"), | |
| gr.Textbox(label="Result") | |
| ], | |
| title="Construction Project Progress Tracker", | |
| description="Upload a photo of your construction site, and the AI will tell you the progress!" | |
| ) | |
| # Launch the Gradio app | |
| if __name__ == "__main__": | |
| iface.launch() |