""" Deterministic symptom inference engine for JAIM. This module adds a lightweight symbolic layer on top of retrieval: - Scores symptom queries against known disorder entries. - Uses keyword overlap, phrase indicators, and retrieval metadata boosts. - Returns a structured diagnosis payload for UI and downstream prompting. """ from __future__ import annotations import re from typing import Any from embed_pdf import extract_entries_from_pdf STOPWORDS = { "a", "an", "and", "are", "as", "at", "be", "been", "by", "for", "from", "has", "have", "i", "in", "is", "it", "its", "my", "of", "on", "or", "that", "the", "their", "there", "these", "they", "this", "to", "was", "were", "with", "without", "very", "so", "just", "really", "also", "but", "do", "does", "did", "will", "would", "can", "could", "should", "not", "no", "some", "all", "any", "much", "many", "lot", "lots", "like", "look", "looks", "looking", "seem", "seems", "get", "gets", "getting", "got", "been", "being", "think", "know", "see", "saw", "noticed", "notice", } SYNONYM_MAP = { # ── Plural / morphological normalization ────────────────── "fruits": "fruit", "leaves": "leaf", "flowers": "flower", "seedlings": "seedling", "seeds": "seed", "trees": "tree", "plants": "plant", "roots": "root", "branches": "branch", "trunks": "trunk", "stems": "stem", "buds": "bud", "crops": "crop", # ── Wound / injury ──────────────────────────────────────── "stunted": "stunt", "wounded": "wound", "wounds": "wound", "wounding": "wound", "injured": "wound", "injuries": "wound", "injury": "wound", "damage": "wound", "damaged": "wound", "broken": "wound", "cutting": "cut", "chopped": "cut", "chopping": "cut", "slashed": "cut", "hacked": "cut", # ── Water-related ───────────────────────────────────────── "waterlogged": "overwater", "overwatering": "overwater", "overwatered": "overwater", "watering": "overwater", "watered": "overwater", "soggy": "overwater", "flooded": "overwater", "flooding": "overwater", "drenched": "overwater", "soaked": "overwater", "waterlog": "overwater", # ── Color / appearance ──────────────────────────────────── "yellowing": "yellow", "yellowed": "yellow", "yellowish": "yellow", "turning yellow": "yellow", "browning": "brown", "browned": "brown", "brownish": "brown", "discolored": "discolor", "discoloring": "discolor", "discoloured": "discolor", "pale": "yellow", "faded": "yellow", # ── Decay / death ───────────────────────────────────────── "oozing": "ooze", "oozes": "ooze", "leaking": "ooze", "seeping": "ooze", "dripping": "ooze", "withering": "wither", "withered": "wither", "wilting": "wilt", "wilted": "wilt", "wilts": "wilt", "drooping": "wilt", "droopy": "wilt", "drooped": "wilt", "sagging": "wilt", "limp": "wilt", "decaying": "decay", "decayed": "decay", "rotting": "rot", "rotted": "rot", "rotten": "rot", "decomposing": "decay", "destroyed": "destroy", "destruction": "destroy", "destroying": "destroy", "dying": "die", "died": "die", "dead": "die", "killing": "die", "killed": "die", # ── Health / condition ──────────────────────────────────── "sick": "unhealthy", "sickly": "unhealthy", "diseased": "unhealthy", "disease": "unhealthy", "infected": "unhealthy", "infection": "unhealthy", "weak": "unhealthy", "weakened": "unhealthy", "unhealthy": "unhealthy", # ── Yield / productivity ────────────────────────────────── "unproductive": "low_yield", "barren": "low_yield", "fruitless": "low_yield", "infertile": "low_yield", "nonproductive": "low_yield", "unfruitful": "low_yield", "no fruit": "low_yield", "no fruits": "low_yield", "not bearing": "low_yield", "not producing": "low_yield", "stopped producing": "low_yield", # ── Smell ───────────────────────────────────────────────── "smelly": "foul_smell", "stinky": "foul_smell", "stinking": "foul_smell", "stinks": "foul_smell", "smells": "foul_smell", "foul": "foul_smell", "stench": "foul_smell", "odor": "foul_smell", "odour": "foul_smell", # ── Growth ──────────────────────────────────────────────── "growing": "grow", "grows": "grow", "growth": "grow", "sprouting": "grow", "sprout": "grow", "germinating": "grow", "germinate": "grow", # ── Shape / deformation ─────────────────────────────────── "bending": "bent", "bends": "bent", "bend": "bent", "crooked": "bent", "twisted": "bent", "twisting": "bent", "curved": "bent", "curving": "bent", "leaning": "bent", "tilting": "bent", "tilted": "bent", # ── Drying ──────────────────────────────────────────────── "drying": "dry", "dried": "dry", "dries": "dry", "parched": "dry", "shriveled": "dry", "shrivelled": "dry", "crispy": "dry", # ── Falling / dropping ──────────────────────────────────── "falling": "fall", "fallen": "fall", "dropping": "fall", "dropped": "fall", "shedding": "fall", "defoliation": "fall", "defoliating": "fall", "losing": "fall", "lost": "fall", # ── Insects / pests ─────────────────────────────────────── "ants": "ant", "insects": "insect", "bugs": "insect", "bug": "insect", "pest": "insect", "pests": "insect", "worms": "worm", "caterpillars": "worm", "caterpillar": "worm", "larvae": "worm", "infestation": "insect", "infested": "insect", # ── Texture / quality ───────────────────────────────────── "tasteless": "bland", "flavorless": "bland", "flavourless": "bland", "no taste": "bland", "hard": "hard", "tough": "hard", "firm": "hard", "mushy": "overripe", "soft": "overripe", "squishy": "overripe", "delayed": "delay", "late": "delay", "slow": "delay", "slowly": "delay", } ENTRY_INDICATORS = { "Vata": [ "bent trunk", "trunk bent", "hard fruit", "not juicy", "yellow leaf", "yellow", "slow defoliation", "loss of flowers", "fall", "bent", "knot", "dry", "hard", "fruit hard", "leaf yellow", "fruit fall", "flower fall", "trunk crooked", "trunk twisted", "fruit not sweet", "fruit less juicy", "arid", "leaf fall", ], "Kapha": [ "delayed fruit", "bland fruit", "bland", "overripe fruit", "overripe", "ooze", "winter", "spring", "delay", "fruit bland", "fruit ooze", "fruit mushy", "fruit soft", "tasteless", "no taste", "fruit rot", "fruit leak", "delayed bearing", "gummy", "gummies", "sap ooze", ], "Pitta": [ "early wither", "leaf wither", "flower decay", "fruit decay", "summer", "wither", "wilt", "decay", "rot", "early", "leaf dry", "leaf wilt", "flower rot", "flower wilt", "leaf brown", "leaf fall early", "flower die", "fruit die", "die early", "wither early", "blight", "scorch", "burn", "heat", ], "Struck by axe etc": [ "wound", "axe", "cut", "dry", "injury", "chop", "slash", "hack", "tree wound", "bark damage", "bark peel", "physical damage", "broken branch", "tree cut", "trunk cut", "branch cut", "tree dry", ], "Faulty Seed": [ "low_yield", "seed", "seed issue", "faulty seed", "not growing", "unproductive", "bad seed", "poor seed", "no fruit", "barren", "fruitless", "not bearing", "not producing", "infertile", "seed fail", "seed problem", "germinate", "sprout fail", "no grow", "stopped producing", ], "Ants": [ "ant", "foul_smell", "foul smell", "bad smell", "smell bad", "stunt", "small leaf", "insect", "worm", "bug", "pest", "caterpillar", "leaf small", "seedling stunt", "fragrance missing", "no fragrance", "stink", "infestation", ], "Excessive watering": [ "overwater", "too much water", "water stress", "indigestion", "root damage", "destroy", "soggy", "flood", "waterlog", "drench", "soak", "root rot", "plant drown", "excess water", "water too much", "gave too much water", "plant die water", "plant sick water", "unhealthy overwater", ], } def _normalize_text(text: str) -> str: text = text.lower() text = re.sub(r"[^a-z0-9\s]", " ", text) text = re.sub(r"\s+", " ", text).strip() return text def _normalize_token(token: str) -> str: token = token.strip().lower() # First pass: direct synonym lookup token = SYNONYM_MAP.get(token, token) # Suffix stripping (simple stemmer) if token.endswith("ation") and len(token) > 6: token = token[:-5] elif token.endswith("tion") and len(token) > 5: token = token[:-4] elif token.endswith("ness") and len(token) > 5: token = token[:-4] elif token.endswith("ment") and len(token) > 5: token = token[:-4] elif token.endswith("ing") and len(token) > 4: token = token[:-3] elif token.endswith("ied") and len(token) > 4: token = token[:-3] + "y" elif token.endswith("ed") and len(token) > 3: token = token[:-2] elif token.endswith("ly") and len(token) > 4: token = token[:-2] elif token.endswith("ies") and len(token) > 4: token = token[:-3] + "y" elif token.endswith("es") and len(token) > 4: token = token[:-2] elif token.endswith("s") and len(token) > 3: token = token[:-1] # Second pass: synonym lookup after stemming token = SYNONYM_MAP.get(token, token) return token def _tokenize(text: str) -> set[str]: normalized = _normalize_text(text) raw_tokens = normalized.split() tokens = set() for token in raw_tokens: if token in STOPWORDS: continue cleaned = _normalize_token(token) if cleaned and cleaned not in STOPWORDS: tokens.add(cleaned) return tokens class SymptomInferenceEngine: """Rule + similarity based inference for plant disorder diagnosis.""" def __init__(self, pdf_path: str = "db.pdf"): self.entries = extract_entries_from_pdf(pdf_path) self._prepared_entries = self._prepare_entries(self.entries) def _prepare_entries(self, entries: list[dict[str, Any]]) -> list[dict[str, Any]]: prepared: list[dict[str, Any]] = [] for idx, entry in enumerate(entries): possible_causes = entry.get("possible_causes", []) if isinstance(possible_causes, list): possible_causes_text = " ".join(possible_causes) else: possible_causes_text = str(possible_causes) full_text = " ".join( [ str(entry.get("disorder", "")), str(entry.get("cause_given", "")), str(entry.get("symptoms", "")), str(entry.get("cause_elaborated", "")), possible_causes_text, ] ) symptom_text = str(entry.get("symptoms", "")) cause = str(entry.get("cause_given", "Unknown")).strip() or "Unknown" prepared.append( { "entry_id": f"entry_{idx}", "cause": cause, "disorder": str(entry.get("disorder", "")).strip() or "Unspecified", "tokens": _tokenize(full_text), "symptom_tokens": _tokenize(symptom_text), "indicators": [i.lower() for i in ENTRY_INDICATORS.get(cause, [])], "raw": entry, } ) return prepared def _phrase_score( self, query_text: str, query_tokens: set[str], indicators: list[str], ) -> tuple[float, list[str]]: if not indicators: return 0.0, [] matched: list[str] = [] soft_total = 0.0 for phrase in indicators: normalized_phrase = _normalize_text(phrase) if normalized_phrase in query_text: matched.append(phrase) soft_total += 1.0 continue phrase_tokens = [_normalize_token(tok) for tok in normalized_phrase.split()] phrase_tokens = [tok for tok in phrase_tokens if tok] if not phrase_tokens: continue token_hits = sum(1 for tok in phrase_tokens if tok in query_tokens) soft_total += token_hits / len(phrase_tokens) if token_hits == len(phrase_tokens): matched.append(phrase) score = soft_total / len(indicators) return min(score, 1.0), matched def _retrieval_boost( self, prepared_entry: dict[str, Any], retrieved_sources: list[dict[str, Any]] | None, ) -> float: if not retrieved_sources: return 0.0 cause = prepared_entry["cause"].lower() disorder = prepared_entry["disorder"].lower() best = 0.0 for source in retrieved_sources: metadata = source.get("metadata", {}) source_cause = str(metadata.get("cause_given", "")).lower() source_disorder = str(metadata.get("disorder", "")).lower() score = float(source.get("score", 0.0)) if source_cause == cause: best = max(best, score) elif disorder != "unspecified" and source_disorder == disorder: best = max(best, score * 0.8) return min(best, 1.0) def infer( self, user_query: str, retrieved_sources: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Return structured diagnosis candidates from query + retrieval evidence.""" query_text = _normalize_text(user_query) query_tokens = _tokenize(user_query) if not query_tokens: return { "engine": "symbolic_v1", "primary_diagnosis": None, "alternatives": [], "note": "No useful symptom tokens found in the query.", } scored_candidates = [] for entry in self._prepared_entries: overlap = query_tokens.intersection(entry["tokens"]) symptom_overlap = query_tokens.intersection(entry["symptom_tokens"]) # Cap denominator so long user descriptions don't dilute good matches. denom = max(min(len(query_tokens), 6), 1) global_token_score = min(len(overlap) / denom, 1.0) symptom_token_score = min(len(symptom_overlap) / denom, 1.0) phrase_score, matched_indicators = self._phrase_score( query_text, query_tokens, entry["indicators"], ) retrieval_score = self._retrieval_boost(entry, retrieved_sources) final_score = ( (0.30 * symptom_token_score) + (0.15 * global_token_score) + (0.35 * phrase_score) + (0.20 * retrieval_score) ) # Calibration bonuses for strong direct evidence. if matched_indicators: final_score += min(0.25, 0.08 * len(matched_indicators)) if phrase_score >= 0.2: final_score += 0.08 if symptom_token_score >= 0.25: final_score += 0.08 if retrieval_score >= 0.5: final_score += 0.06 if global_token_score >= 0.3: final_score += 0.04 scored_candidates.append( { "entry_id": entry["entry_id"], "cause": entry["cause"], "disorder": entry["disorder"], "score": round(min(final_score, 1.0), 4), "matched_terms": sorted(overlap), "matched_symptom_terms": sorted(symptom_overlap), "matched_indicators": matched_indicators, "treatment_material": entry["raw"].get("treatment_material", "N/A"), "possible_causes": entry["raw"].get("possible_causes", []), "symptoms": entry["raw"].get("symptoms", "N/A"), } ) scored_candidates.sort(key=lambda item: item["score"], reverse=True) top = scored_candidates[0] alternatives = scored_candidates[1:3] confidence_label = "low" if top["score"] >= 0.65: confidence_label = "high" elif top["score"] >= 0.35: confidence_label = "medium" rationale = ( f"Matched indicators: {', '.join(top['matched_indicators']) or 'none'}; " f"Matched symptom terms: {', '.join(top['matched_symptom_terms'][:8]) or 'none'}." ) primary = { "entry_id": top["entry_id"], "disorder": top["disorder"], "cause": top["cause"], "confidence_score": top["score"], "confidence_label": confidence_label, "rationale": rationale, "matched_indicators": top["matched_indicators"], "treatment_material": top["treatment_material"], "possible_causes": top["possible_causes"], "reference_symptoms": top["symptoms"], } return { "engine": "symbolic_v1", "primary_diagnosis": primary, "alternatives": alternatives, }