# """ # service.py # LeafDoctor AI # Business Service Layer # Part 1 # """ # from __future__ import annotations # import shutil # import uuid # from datetime import datetime # from pathlib import Path # from sqlalchemy import func # from sqlalchemy.orm import Session # from config import settings # from gemini import GeminiClient, DiagnosisResult # from models import ( # DiagnosisHistory, # ProtectionLabel, # TreatmentMapping, # ) # class DiagnosisService: # def __init__(self, db: Session): # self.db = db # self.gemini = GeminiClient.instance() # # -------------------------------------------------- # # Image # # -------------------------------------------------- # def _generate_filename( # self, # extension: str, # ) -> str: # timestamp = datetime.now().strftime( # "%Y%m%d%H%M%S" # ) # random_id = uuid.uuid4().hex[:8] # return ( # f"{timestamp}_" # f"{random_id}" # f"{extension}" # ) # def save_image( # self, # image_path: str | Path, # ) -> Path: # image_path = Path(image_path) # if not image_path.exists(): # raise FileNotFoundError( # image_path # ) # extension = image_path.suffix.lower() # filename = self._generate_filename( # extension # ) # destination = ( # settings.upload_path / # filename # ) # destination.parent.mkdir( # parents=True, # exist_ok=True, # ) # shutil.copy2( # image_path, # destination, # ) # return destination # # -------------------------------------------------- # # Protection Label # # -------------------------------------------------- # def get_label( # self, # label: str, # ) -> ProtectionLabel: # obj = ( # self.db.query( # ProtectionLabel # ) # .filter( # func.lower( # ProtectionLabel.label_name_en # ) == label.lower() # ) # .first() # ) # if obj is None: # raise ValueError( # f"Unknown label '{label}'" # ) # return obj # # -------------------------------------------------- # # Treatment # # -------------------------------------------------- # def get_treatments( # self, # class_id: int, # crop: str, # ) -> list[TreatmentMapping]: # return ( # self.db.query( # TreatmentMapping # ) # .filter( # TreatmentMapping.class_id == class_id, # func.lower( # TreatmentMapping.crop_en # ) == crop.lower(), # ) # ) # # -------------------------------------------------- # # History # # -------------------------------------------------- # def save_history( # self, # image_path: str, # diagnosis: DiagnosisResult, # treatment: TreatmentMapping | None, # class_id: int, # ) -> DiagnosisHistory: # history = DiagnosisHistory( # image_path=image_path, # crop_detected=diagnosis.crop, # predicted_label=diagnosis.label, # class_id=class_id, # confidence=diagnosis.confidence, # treatment_id=( # treatment.id # if treatment # else None # ), # gemini_model=diagnosis.model, # raw_response=diagnosis.raw_response, # ) # try: # self.db.add(history) # self.db.commit() # self.db.refresh(history) # return history # except Exception: # self.db.rollback() # raise # # -------------------------------------------------- # # Helpers # # -------------------------------------------------- # @staticmethod # def best_treatment( # treatments: list[TreatmentMapping], # ) -> TreatmentMapping | None: # if not treatments: # return None # if not treatments: # return None # return sorted( # treatments, # key=lambda x: ( # x.chemical is None, # x.dosage is None, # x.sms_en is None, # ), # )[0] # # -------------------------------------------------- # # Diagnosis Pipeline # # -------------------------------------------------- # def diagnose( # self, # image_path: str | Path, # ) -> dict: # # --------------------------------------- # # Save uploaded image # # --------------------------------------- # stored_image = self.save_image( # image_path # ) # # --------------------------------------- # # Gemini Diagnosis # # --------------------------------------- # diagnosis = self.gemini.diagnose( # stored_image # ) # # --------------------------------------- # # Master Label # # --------------------------------------- # label = self.get_label( # diagnosis.label # ) # # --------------------------------------- # # Treatments # # --------------------------------------- # treatments = self.get_treatments( # class_id=label.class_id, # crop=diagnosis.crop, # ) # treatment = self.best_treatment( # treatments # ) # # --------------------------------------- # # Save History # # --------------------------------------- # history = self.save_history( # image_path=str(stored_image), # diagnosis=diagnosis, # treatment=treatment, # class_id=label.class_id, # ) # # --------------------------------------- # # Response # # --------------------------------------- # return self._build_response( # history=history, # diagnosis=diagnosis, # label=label, # treatments=treatments, # stored_image=stored_image, # ) # # -------------------------------------------------- # # Response Builder # # -------------------------------------------------- # def _build_response( # self, # history: DiagnosisHistory, # diagnosis: DiagnosisResult, # label: ProtectionLabel, # treatments: list[TreatmentMapping], # stored_image: Path, # ) -> dict: # recommendations = [] # for item in treatments: # recommendations.append( # { # "id": item.id, # "crop": { # "en": item.crop_en, # "bn": item.crop_bn, # }, # "problem_type": { # "en": item.problem_type_en, # "bn": item.problem_type_bn, # }, # "chemical": item.chemical, # "chemical_type": item.chemical_type, # "dosage": item.dosage, # "sms": { # "en": item.sms_en, # "bn": item.sms_bn, # }, # } # ) # return { # "success": True, # "history_id": history.id, # "image": str(stored_image), # "diagnosis": { # "class_id": label.class_id, # "confidence": diagnosis.confidence, # "crop": { # "en": diagnosis.crop, # "bn": ( # treatments[0].crop_bn # if treatments # else None # ), # }, # "problem": { # "en": label.label_name_en, # "bn": label.label_name_bn, # }, # "category": { # "en": label.primary_category_en, # "bn": label.primary_category_bn, # }, # "gemini_model": diagnosis.model, # }, # "recommendations": recommendations, # } # # -------------------------------------------------- # # Utility Methods # # -------------------------------------------------- # def get_history( # self, # history_id: int, # ) -> DiagnosisHistory | None: # return ( # self.db.query( # DiagnosisHistory # ) # .filter( # DiagnosisHistory.id == history_id # ) # .first() # ) # def get_histories( # self, # limit: int = 50, # ) -> list[DiagnosisHistory]: # return ( # self.db.query( # DiagnosisHistory # ) # .order_by( # DiagnosisHistory.created_at.desc() # ) # .limit(limit) # .all() # ) # def delete_history( # self, # history_id: int, # ) -> bool: # history = self.get_history( # history_id # ) # if history is None: # return False # try: # image_path = Path( # history.image_path # ) # if image_path.exists(): # image_path.unlink() # except Exception: # pass # self.db.delete(history) # self.db.commit() # return True # # -------------------------------------------------- # # Health Check # # -------------------------------------------------- # def health(self): # database = "connected" # gemini = "connected" # try: # self.db.execute("SELECT 1") # except Exception: # database = "disconnected" # try: # self.gemini.test_connection() # except Exception: # gemini = "disconnected" # return { # "success": database == "connected" # and gemini == "connected", # "database": database, # "gemini": gemini, # "model": self.gemini.model, # } # # -------------------------------------------------- # # Statistics # # -------------------------------------------------- # def statistics(self) -> dict: # total_labels = ( # self.db.query( # ProtectionLabel # ) # .count() # ) # total_treatments = ( # self.db.query( # TreatmentMapping # ) # .count() # ) # total_diagnosis = ( # self.db.query( # DiagnosisHistory # ) # .count() # ) # return { # "labels": total_labels, # "treatments": total_treatments, # "diagnosis_history": total_diagnosis, # } """ service.py LeafDoctor AI Business Service Layer """ from __future__ import annotations import shutil import uuid from datetime import datetime from pathlib import Path from sqlalchemy import func from sqlalchemy.orm import Session from config import settings from gemini import GeminiClient, DiagnosisResult from models import ( DiagnosisHistory, ProtectionLabel, TreatmentMapping, ) class DiagnosisService: def __init__(self, db: Session): self.db = db self.gemini = GeminiClient.instance() # -------------------------------------------------- # Image # -------------------------------------------------- def _generate_filename( self, extension: str, ) -> str: timestamp = datetime.now().strftime( "%Y%m%d%H%M%S" ) random_id = uuid.uuid4().hex[:8] return ( f"{timestamp}_" f"{random_id}" f"{extension}" ) def save_image( self, image_path: str | Path, ) -> Path: image_path = Path(image_path) if not image_path.exists(): raise FileNotFoundError( image_path ) extension = image_path.suffix.lower() filename = self._generate_filename( extension ) destination = ( settings.upload_path / filename ) destination.parent.mkdir( parents=True, exist_ok=True, ) shutil.copy2( image_path, destination, ) return destination # -------------------------------------------------- # Protection Label # -------------------------------------------------- def get_label( self, label: str, ) -> ProtectionLabel: obj = ( self.db.query( ProtectionLabel ) .filter( func.lower( ProtectionLabel.label_name_en ) == label.lower() ) .first() ) if obj is None: raise ValueError( f"Unknown label '{label}'" ) return obj # -------------------------------------------------- # Treatment # -------------------------------------------------- def get_treatments( self, class_id: int, crop: str, ) -> list[TreatmentMapping]: return ( self.db.query( TreatmentMapping ) .filter( TreatmentMapping.class_id == class_id, func.lower( TreatmentMapping.crop_en ) == crop.lower(), ) .all() ) # -------------------------------------------------- # History # -------------------------------------------------- def save_history( self, image_path: str, diagnosis: DiagnosisResult, treatment: TreatmentMapping | None, class_id: int, user_id: str | None = None, user_ph: str | None = None, ) -> DiagnosisHistory: history = DiagnosisHistory( user_id=user_id, user_ph=user_ph, image_path=image_path, crop_detected=diagnosis.crop, predicted_label=diagnosis.label, class_id=class_id, confidence=diagnosis.confidence, treatment_id=( treatment.id if treatment else None ), gemini_model=diagnosis.model, raw_response=diagnosis.raw_response, ) try: self.db.add(history) self.db.commit() self.db.refresh(history) return history except Exception: self.db.rollback() raise # -------------------------------------------------- # Helpers # -------------------------------------------------- @staticmethod def best_treatment( treatments: list[TreatmentMapping], ) -> TreatmentMapping | None: if not treatments: return None return sorted( treatments, key=lambda x: ( x.chemical is None, x.dosage is None, x.sms_en is None, ), )[0] # -------------------------------------------------- # Diagnosis Pipeline # -------------------------------------------------- def diagnose( self, image_path: str | Path, user_id: str | None = None, user_ph: str | None = None, ) -> dict: # --------------------------------------- # Save uploaded image # --------------------------------------- stored_image = self.save_image( image_path ) # --------------------------------------- # Gemini Diagnosis # --------------------------------------- diagnosis = self.gemini.diagnose( stored_image ) # --------------------------------------- # Master Label # --------------------------------------- label = self.get_label( diagnosis.label ) # --------------------------------------- # Treatments # --------------------------------------- treatments = self.get_treatments( class_id=label.class_id, crop=diagnosis.crop, ) treatment = self.best_treatment( treatments ) # --------------------------------------- # Save History # --------------------------------------- history = self.save_history( image_path=str(stored_image), diagnosis=diagnosis, treatment=treatment, class_id=label.class_id, user_id=user_id, user_ph=user_ph, ) # --------------------------------------- # Response # --------------------------------------- return self._build_response( history=history, diagnosis=diagnosis, label=label, treatments=treatments, stored_image=stored_image, ) # -------------------------------------------------- # Response Builder # -------------------------------------------------- def _build_response( self, history: DiagnosisHistory, diagnosis: DiagnosisResult, label: ProtectionLabel, treatments: list[TreatmentMapping], stored_image: Path, ) -> dict: recommendations = [] for item in treatments: recommendations.append( { "id": item.id, "crop": { "en": item.crop_en, "bn": item.crop_bn, }, "problem_type": { "en": item.problem_type_en, "bn": item.problem_type_bn, }, "chemical": item.chemical, "chemical_type": item.chemical_type, "dosage": item.dosage, "sms": { "en": item.sms_en, "bn": item.sms_bn, }, } ) return { "success": True, "history_id": history.id, "user_id": history.user_id, "user_ph": history.user_ph, "image": str(stored_image), "diagnosis": { "class_id": label.class_id, "confidence": diagnosis.confidence, "crop": { "en": diagnosis.crop, "bn": ( treatments[0].crop_bn if treatments else None ), }, "problem": { "en": label.label_name_en, "bn": label.label_name_bn, }, "category": { "en": label.primary_category_en, "bn": label.primary_category_bn, }, "gemini_model": diagnosis.model, }, "recommendations": recommendations, } # -------------------------------------------------- # Utility Methods # -------------------------------------------------- def get_history( self, history_id: int, ) -> DiagnosisHistory | None: return ( self.db.query( DiagnosisHistory ) .filter( DiagnosisHistory.id == history_id ) .first() ) def get_histories( self, limit: int = 50, ) -> list[DiagnosisHistory]: return ( self.db.query( DiagnosisHistory ) .order_by( DiagnosisHistory.created_at.desc() ) .limit(limit) .all() ) def delete_history( self, history_id: int, ) -> bool: history = self.get_history( history_id ) if history is None: return False try: image_path = Path( history.image_path ) if image_path.exists(): image_path.unlink() except Exception: pass self.db.delete(history) self.db.commit() return True # -------------------------------------------------- # Health Check # -------------------------------------------------- def health(self): database = "connected" gemini = "connected" try: self.db.execute("SELECT 1") except Exception: database = "disconnected" try: self.gemini.test_connection() except Exception: gemini = "disconnected" return { "success": database == "connected" and gemini == "connected", "database": database, "gemini": gemini, "model": self.gemini.model, } # -------------------------------------------------- # Statistics # -------------------------------------------------- def statistics(self) -> dict: total_labels = ( self.db.query( ProtectionLabel ) .count() ) total_treatments = ( self.db.query( TreatmentMapping ) .count() ) total_diagnosis = ( self.db.query( DiagnosisHistory ) .count() ) return { "labels": total_labels, "treatments": total_treatments, "diagnosis_history": total_diagnosis, } # -------------------------------------------------- # History using user_id or user_ph # -------------------------------------------------- def get_user_histories( self, user_id: str | None = None, user_ph: str | None = None, limit: int = 50, ) -> list[DiagnosisHistory]: if not user_id and not user_ph: return [] query = self.db.query(DiagnosisHistory) # Filter dynamically based on provided parameters if user_id and user_ph: query = query.filter( (DiagnosisHistory.user_id == user_id) | (DiagnosisHistory.user_ph == user_ph) ) elif user_id: query = query.filter(DiagnosisHistory.user_id == user_id) elif user_ph: query = query.filter(DiagnosisHistory.user_ph == user_ph) return ( query.order_by(DiagnosisHistory.created_at.desc()) .limit(limit) .all() )