Plant Disease Diagnosis with Ontologies

Building an Ontology-Driven Diagnostic System

This guide demonstrates how to build a comprehensive plant disease diagnosis system using ontologies, combining formal knowledge representation with machine learning for accurate and explainable disease identification.

System Architecture

graph TD
    A[Input: Plant Images & Symptoms] --> B[Feature Extraction]
    B --> C[Ontology-Based Reasoning]
    C --> D[Disease Diagnosis]
    D --> E[Treatment Recommendations]
    E --> F[Output: Diagnosis Report]
    
    G[Plant Disease Ontology] --> C
    H[Knowledge Base] --> C
    
    style G fill:#f9f,stroke:#333
    style H fill:#9cf,stroke:#333

Core Components

1. Ontology Design

from rdflib import Graph, Namespace, Literal
from rdflib.namespace import RDF, RDFS, OWL, XSD

# Define namespaces
PLANT = Namespace("http://example.org/plant-ontology#")
SYMPTOM = Namespace("http://example.org/symptom-ontology#")
DISEASE = Namespace("http://example.org/disease-ontology#")

# Initialize graph
g = Graph()

# Define classes
g.add((PLANT.Plant, RDF.type, OWL.Class))
g.add((PLANT.Leaf, RDF.type, OWL.Class))
g.add((PLANT.Leaf, RDFS.subClassOf, PLANT.PlantPart))
g.add((SYMPTOM.Yellowing, RDF.type, OWL.Class))
g.add((SYMPTOM.Spots, RDF.type, OWL.Class))

# Define object properties
g.add((PLANT.hasSymptom, RDF.type, OWL.ObjectProperty))
g.add((PLANT.hasSymptom, RDFS.domain, PLANT.Plant))
g.add((PLANT.hasSymptom, RDFS.range, SYMPTOM.PlantSymptom))

# Define data properties
g.add((PLANT.hasSeverity, RDF.type, OWL.DatatypeProperty))
g.add((PLANT.hasSeverity, RDFS.domain, PLANT.Plant))
g.add((PLANT.hasSeverity, RDFS.range, XSD.integer))

# Save ontology
g.serialize("plant_disease_ontology.ttl", format="turtle")

2. Symptom to Disease Mapping

import pandas as pd
from typing import List, Dict

class DiseaseDiagnoser:
    def __init__(self, ontology_path: str):
        self.ontology = Graph()
        self.ontology.parse(ontology_path, format="turtle")
        self.symptom_to_diseases = self._build_symptom_index()
    
    def _build_symptom_index(self) -> Dict[str, List[str]]:
        """Build an index of symptoms to diseases."""
        query = """
        PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
        PREFIX plant: <http://example.org/plant-ontology#>
        
        SELECT ?symptom ?disease
        WHERE {
            ?disease rdf:type plant:Disease ;
                     plant:hasSymptom ?symptom .
        }
        """
        results = self.ontology.query(query)
        
        index = {}
        for row in results:
            symptom = str(row.symptom).split("#")[-1]
            disease = str(row.disease).split("#")[-1]
            if symptom not in index:
                index[symptom] = []
            index[symptom].append(disease)
            
        return index
    
    def diagnose(self, symptoms: List[str]) -> pd.DataFrame:
        """Diagnose potential diseases based on symptoms."""
        disease_scores = {}
        
        for symptom in symptoms:
            for disease in self.symptom_to_diseases.get(symptom, []):
                disease_scores[disease] = disease_scores.get(disease, 0) + 1
        
        # Convert to DataFrame for better visualization
        df = pd.DataFrame({
            'Disease': list(disease_scores.keys()),
            'Match Score': list(disease_scores.values())
        }).sort_values('Match Score', ascending=False)
        
        return df

Integration with ML Models

1. Image-Based Symptom Detection

import torch
from torchvision import models, transforms
from PIL import Image

class SymptomDetector:
    def __init__(self, model_path: str):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = models.resnet50(pretrained=False)
        num_ftrs = self.model.fc.in_features
        self.model.fc = torch.nn.Linear(num_ftrs, 10)  # 10 symptom classes
        self.model.load_state_dict(torch.load(model_path, map_location=self.device))
        self.model.eval()
        
        self.transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
        ])
        
        self.classes = [
            'healthy', 'yellowing', 'spots', 'wilting', 'mold',
            'blight', 'rust', 'mildew', 'necrosis', 'stunting'
        ]
    
    def detect(self, image_path: str, threshold: float = 0.5) -> List[Dict]:
        """Detect symptoms in a plant image."""
        image = Image.open(image_path).convert('RGB')
        image_tensor = self.transform(image).unsqueeze(0).to(self.device)
        
        with torch.no_grad():
            outputs = torch.sigmoid(self.model(image_tensor))
        
        results = []
        for i, score in enumerate(outputs.squeeze().cpu().numpy()):
            if score > threshold:
                results.append({
                    'symptom': self.classes[i],
                    'confidence': float(score)
                })
        
        return results

End-to-End Diagnosis Pipeline

class PlantDiseaseDiagnosisSystem:
    def __init__(self, ontology_path: str, model_path: str):
        self.diagnoser = DiseaseDiagnoser(ontology_path)
        self.detector = SymptomDetector(model_path)
        self.knowledge_base = self._load_knowledge_base()
    
    def _load_knowledge_base(self) -> Dict:
        """Load treatment and prevention knowledge."""
        return {
            'powdery_mildew': {
                'treatment': 'Apply sulfur or potassium bicarbonate-based fungicides',
                'prevention': 'Ensure good air circulation, avoid overhead watering',
                'severity': 'moderate'
            },
            'late_blight': {
                'treatment': 'Apply copper-based fungicides, remove infected plants',
                'prevention': 'Use disease-free seeds, practice crop rotation',
                'severity': 'high'
            },
            # Add more diseases and treatments
        }
    
    def diagnose(self, image_path: str, additional_symptoms: List[str] = None) -> Dict:
        """Perform end-to-end diagnosis."""
        # Detect symptoms from image
        detected_symptoms = self.detector.detect(image_path)
        symptom_names = [s['symptom'] for s in detected_symptoms]
        
        # Add any additional symptoms provided by user
        if additional_symptoms:
            symptom_names.extend(additional_symptoms)
        
        # Get potential diseases
        diagnosis = self.diagnoser.diagnose(symptom_names)
        
        # Add treatment information
        results = []
        for _, row in diagnosis.iterrows():
            disease = row['Disease']
            if disease in self.knowledge_base:
                result = {
                    'disease': disease,
                    'match_score': row['Match Score'],
                    **self.knowledge_base[disease]
                }
                results.append(result)
        
        return {
            'detected_symptoms': symptom_names,
            'possible_diagnoses': results
        }

Evaluation

1. Performance Metrics

from sklearn.metrics import precision_score, recall_score, f1_score
import numpy as np

class Evaluator:
    @staticmethod
    def evaluate(y_true: List[str], y_pred: List[str]) -> Dict:
        """Evaluate diagnosis performance."""
        # Convert to binary vectors
        all_labels = sorted(list(set(y_true + y_pred)))
        y_true_bin = [[1 if label in true else 0 for label in all_labels] 
                      for true in y_true]
        y_pred_bin = [[1 if label in pred else 0 for label in all_labels]
                     for pred in y_pred]
        
        # Calculate metrics
        precision = precision_score(y_true_bin, y_pred_bin, average='micro')
        recall = recall_score(y_true_bin, y_pred_bin, average='micro')
        f1 = f1_score(y_true_bin, y_pred_bin, average='micro')
        
        return {
            'precision': precision,
            'recall': recall,
            'f1_score': f1
        }
    
    @staticmethod
    def confusion_matrix(y_true: List[str], y_pred: List[str]) -> pd.DataFrame:
        """Generate confusion matrix."""
        from sklearn.metrics import confusion_matrix
        
        # Get all unique labels
        labels = sorted(list(set(y_true + y_pred)))
        cm = confusion_matrix(y_true, y_pred, labels=labels)
        
        return pd.DataFrame(cm, index=labels, columns=labels)

Deployment

1. FastAPI Web Service

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI(title="Plant Disease Diagnosis API")

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize system
system = PlantDiseaseDiagnosisSystem(
    ontology_path="plant_disease_ontology.ttl",
    model_path="symptom_detector.pth"
)

@app.post("/diagnose")
async def diagnose_plant(
    image: UploadFile = File(...),
    symptoms: str = ""
):
    try:
        # Save uploaded image
        image_path = f"temp_{image.filename}"
        with open(image_path, "wb") as buffer:
            buffer.write(await image.read())
        
        # Process symptoms
        additional_symptoms = [s.strip() for s in symptoms.split(",") if s.strip()]
        
        # Get diagnosis
        result = system.diagnose(image_path, additional_symptoms)
        
        # Clean up
        import os
        os.remove(image_path)
        
        return result
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Next Steps

1. Expand Ontology: Add more plant species, diseases, and symptoms
2. Improve Models: Train on larger datasets for better accuracy
3. Mobile App: Create a mobile interface for field diagnosis
4. Continuous Learning: Update the system with new cases and treatments

References

• Plant Disease Ontology
• FAO Plant Protection
• APS Plant Disease Lessons