import streamlit as st import torch from PIL import Image import numpy as np from transformers import ViTFeatureExtractor, ViTForImageClassification from sentence_transformers import SentenceTransformer import matplotlib.pyplot as plt import logging import faiss from typing import List, Dict from datetime import datetime from groq import Groq import os from functools import lru_cache # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RAGSystem: def __init__(self): # Load models only when needed self._embedding_model = None self._vector_store = None self._knowledge_base = None @property def embedding_model(self): if self._embedding_model is None: self._embedding_model = SentenceTransformer('all-MiniLM-L6-v2') return self._embedding_model @property def knowledge_base(self): if self._knowledge_base is None: self._knowledge_base = self.load_knowledge_base() return self._knowledge_base @property def vector_store(self): if self._vector_store is None: self._vector_store = self.create_vector_store() return self._vector_store @staticmethod @lru_cache(maxsize=1) # Cache the knowledge base def load_knowledge_base() -> List[Dict]: """Load and preprocess knowledge base""" kb = { "spalling": [ { "severity": "Critical", "description": "Severe concrete spalling with exposed reinforcement", "repair_method": "Remove deteriorated concrete, clean reinforcement", "immediate_action": "Evacuate area, install support", "prevention": "Regular inspections, waterproofing" } ], "structural_cracks": [ { "severity": "High", "description": "Active structural cracks >5mm width", "repair_method": "Structural analysis, epoxy injection", "immediate_action": "Install crack monitors", "prevention": "Regular monitoring, load management" } ], "surface_deterioration": [ { "severity": "Medium", "description": "Surface scaling and deterioration", "repair_method": "Surface preparation, patch repair", "immediate_action": "Document extent, plan repairs", "prevention": "Surface sealers, proper drainage" } ], "corrosion": [ { "severity": "High", "description": "Corrosion of reinforcement leading to cracks", "repair_method": "Remove rust, apply inhibitors", "immediate_action": "Isolate affected area", "prevention": "Anti-corrosion coatings, proper drainage" } ], "efflorescence": [ { "severity": "Low", "description": "White powder deposits on concrete surfaces", "repair_method": "Surface cleaning, sealant application", "immediate_action": "Identify moisture source", "prevention": "Improve waterproofing, reduce moisture ingress" } ], "delamination": [ { "severity": "Medium", "description": "Separation of layers in concrete", "repair_method": "Resurface or replace delaminated sections", "immediate_action": "Inspect bonding layers", "prevention": "Proper curing and bonding agents" } ], "honeycombing": [ { "severity": "Medium", "description": "Voids in concrete caused by improper compaction", "repair_method": "Grout injection, patch repair", "immediate_action": "Assess structural impact", "prevention": "Proper vibration during pouring" } ], "water_leakage": [ { "severity": "High", "description": "Water ingress through cracks or joints", "repair_method": "Injection grouting, waterproofing membranes", "immediate_action": "Stop water flow, apply sealants", "prevention": "Drainage systems, joint sealing" } ], "settlement_cracks": [ { "severity": "High", "description": "Cracks due to uneven foundation settlement", "repair_method": "Foundation underpinning, grouting", "immediate_action": "Monitor movement, stabilize foundation", "prevention": "Soil compaction, proper foundation design" } ], "shrinkage_cracks": [ { "severity": "Low", "description": "Minor cracks caused by shrinkage during curing", "repair_method": "Sealant application, surface repairs", "immediate_action": "Monitor cracks", "prevention": "Proper curing and moisture control" } ] } documents = [] for category, items in kb.items(): for item in items: doc_text = f"Category: {category}\n" for key, value in item.items(): doc_text += f"{key}: {value}\n" documents.append({"text": doc_text, "metadata": {"category": category}}) return documents def create_vector_store(self): """Create FAISS vector store""" texts = [doc["text"] for doc in self.knowledge_base] embeddings = self.embedding_model.encode(texts) dimension = embeddings.shape[1] index = faiss.IndexFlatL2(dimension) index.add(np.array(embeddings).astype('float32')) return index @lru_cache(maxsize=32) # Cache recent query results def get_relevant_context(self, query: str, k: int = 2) -> str: """Retrieve relevant context based on query""" try: query_embedding = self.embedding_model.encode([query]) D, I = self.vector_store.search(np.array(query_embedding).astype('float32'), k) context = "\n\n".join([self.knowledge_base[i]["text"] for i in I[0]]) return context except Exception as e: logger.error(f"Error retrieving context: {e}") return "" class ImageAnalyzer: def __init__(self, model_name="microsoft/swin-base-patch4-window7-224-in22k"): self.device = "cpu" self.defect_classes = ["spalling", "structural_cracks", "surface_deterioration"] self.model_name = model_name self._model = None self._feature_extractor = None @property def model(self): if self._model is None: self._model = self._load_model() return self._model @property def feature_extractor(self): if self._feature_extractor is None: self._feature_extractor = self._load_feature_extractor() return self._feature_extractor def _load_feature_extractor(self): """Load the appropriate feature extractor based on model type""" try: if "swin" in self.model_name: from transformers import AutoFeatureExtractor return AutoFeatureExtractor.from_pretrained(self.model_name) elif "convnext" in self.model_name: from transformers import ConvNextFeatureExtractor return ConvNextFeatureExtractor.from_pretrained(self.model_name) else: from transformers import ViTFeatureExtractor return ViTFeatureExtractor.from_pretrained(self.model_name) except Exception as e: logger.error(f"Feature extractor initialization error: {e}") return None def _load_model(self): try: if "swin" in self.model_name: from transformers import SwinForImageClassification model = SwinForImageClassification.from_pretrained( self.model_name, num_labels=len(self.defect_classes), ignore_mismatched_sizes=True ) elif "convnext" in self.model_name: from transformers import ConvNextForImageClassification model = ConvNextForImageClassification.from_pretrained( self.model_name, num_labels=len(self.defect_classes), ignore_mismatched_sizes=True ) else: from transformers import ViTForImageClassification model = ViTForImageClassification.from_pretrained( self.model_name, num_labels=len(self.defect_classes), ignore_mismatched_sizes=True ) model = model.to(self.device) # Reinitialize the classifier layer with torch.no_grad(): if hasattr(model, 'classifier'): in_features = model.classifier.in_features model.classifier = torch.nn.Linear(in_features, len(self.defect_classes)) elif hasattr(model, 'head'): in_features = model.head.in_features model.head = torch.nn.Linear(in_features, len(self.defect_classes)) return model except Exception as e: logger.error(f"Model initialization error: {e}") return None def preprocess_image(self, image_bytes): """Preprocess image for model input""" return _cached_preprocess_image(image_bytes, self.model_name) def analyze_image(self, image): """Analyze image for defects""" try: if self.model is None: raise ValueError("Model not properly initialized") inputs = self.feature_extractor( images=image, return_tensors="pt" ) inputs = {k: v.to(self.device) for k, v in inputs.items()} with torch.no_grad(): outputs = self.model(**inputs) probs = torch.nn.functional.softmax(outputs.logits, dim=1)[0] confidence_threshold = 0.3 results = { self.defect_classes[i]: float(probs[i]) for i in range(len(self.defect_classes)) if float(probs[i]) > confidence_threshold } if not results: max_idx = torch.argmax(probs) results = {self.defect_classes[int(max_idx)]: float(probs[max_idx])} return results except Exception as e: logger.error(f"Analysis error: {str(e)}") return None @st.cache_data def _cached_preprocess_image(image_bytes, model_name): """Cached version of image preprocessing""" try: image = Image.open(image_bytes) if image.mode != 'RGB': image = image.convert('RGB') # Adjust size based on model requirements if "convnext" in model_name: width, height = 384, 384 else: width, height = 224, 224 image = image.resize((width, height), Image.Resampling.LANCZOS) return image except Exception as e: logger.error(f"Image preprocessing error: {e}") return None @st.cache_data def get_groq_response(query: str, context: str) -> str: """Get response from Groq LLM with caching""" try: if not os.getenv("GROQ_API_KEY"): return "Error: Groq API key not configured" client = Groq(api_key=os.getenv("GROQ_API_KEY")) prompt = f"""Based on the following context about construction defects, answer the question. Context: {context} Question: {query} Provide a detailed answer based on the given context.""" response = client.chat.completions.create( messages=[ { "role": "system", "content": "You are a construction defect analysis expert." }, { "role": "user", "content": prompt } ], model="llama-3.3-70b-versatile", temperature=0.7, ) return response.choices[0].message.content except Exception as e: logger.error(f"Groq API error: {e}", exc_info=True) return f"Error: Unable to get response from AI model. Exception: {str(e)}" def main(): st.set_page_config( page_title="Smart Construction Defect Analyzer", page_icon="🏗️", layout="wide" ) st.title("🏗️ Smart Construction Defect Analyzer") # Initialize systems in session state if not present if 'analyzer' not in st.session_state: st.session_state.analyzer = ImageAnalyzer() if 'rag_system' not in st.session_state: st.session_state.rag_system = RAGSystem() col1, col2 = st.columns([1, 1]) with col1: st.subheader("Image Analysis") uploaded_file = st.file_uploader( "Upload a construction image for analysis", type=["jpg", "jpeg", "png"], key="image_uploader" # Add key for proper state management ) if uploaded_file is not None: try: # Create a placeholder for the image image_placeholder = st.empty() # Process image with progress indicator with st.spinner('Processing image...'): processed_image = st.session_state.analyzer.preprocess_image(uploaded_file) if processed_image: image_placeholder.image(processed_image, caption='Uploaded Image', use_container_width=True) # Analyze image with progress bar progress_bar = st.progress(0) with st.spinner('Analyzing defects...'): results = st.session_state.analyzer.analyze_image(processed_image) progress_bar.progress(100) if results: st.success('Analysis complete!') # Display results st.subheader("Detected Defects") fig, ax = plt.subplots(figsize=(8, 4)) defects = list(results.keys()) probs = list(results.values()) ax.barh(defects, probs) ax.set_xlim(0, 1) plt.tight_layout() st.pyplot(fig) most_likely_defect = max(results.items(), key=lambda x: x[1])[0] st.info(f"Most likely defect: {most_likely_defect}") else: st.warning("No defects detected or analysis failed. Please try another image.") else: st.error("Failed to process image. Please try another one.") except Exception as e: st.error(f"Error processing image: {str(e)}") logger.error(f"Process error: {e}") with col2: st.subheader("Ask About Defects") user_query = st.text_input( "Ask a question about the defects or repairs:", help="Example: What are the repair methods for spalling?" ) if user_query: with st.spinner('Getting answer...'): # Get context from RAG system context = st.session_state.rag_system.get_relevant_context(user_query) if context: # Get response from Groq response = get_groq_response(user_query, context) if not response.startswith("Error"): st.write("Answer:") st.markdown(response) else: st.error(response) with st.expander("View retrieved information"): st.text(context) else: st.error("Could not find relevant information. Please try rephrasing your question.") with st.sidebar: st.header("About") st.write(""" This tool helps analyze construction defects in images and provides information about repair methods and best practices. Features: - Image analysis for defect detection - Information lookup for repair methods - Expert AI responses to your questions """) # Display API status if os.getenv("GROQ_API_KEY"): st.success("Groq API: Connected") else: st.error("Groq API: Not configured") # Add settings section st.subheader("Settings") if st.button("Clear Cache"): st.cache_data.clear() st.success("Cache cleared!") if __name__ == "__main__": main()