damage_type_classifier.py

"""

Agent 2 - Visual Damage-Type Classifier

------------------------------------------

Takes the flagged region crop from Agent 1 (plus its change_score and

reliability signals) and identifies WHICH KIND of damage it is, using a

Vision-Language Model (Gemini, same free-tier choice as Agent 3) - zero

training, zero fine-tuning, same philosophy as the rest of Module 2.



Classifies into exactly one of the 9 shared taxonomy categories:

  screen_display, structural_body, port_connector, missing_component,

  liquid_moisture, input_control, power_boot_failure, cosmetic_wear,

  other_unclassified



Self-consistency: calls the VLM multiple times on the same crop and checks

agreement - if the calls disagree on damage_type, that's flagged for human

review rather than silently picking one guess. Same honest-about-uncertainty

pattern used throughout Agent 1.

"""

import os
import json
import re
from collections import Counter
from dataclasses import dataclass, field

from google import genai
from google.genai import types


VLM_MODEL = "gemini-3.5-flash"
SELF_CONSISTENCY_CALLS = 3          # how many times to classify the same crop
AGREEMENT_THRESHOLD = 0.6           # fraction of calls that must agree on damage_type
                                     # to trust it without flagging for human review

# The shared taxonomy - must match damage_knowledge_base.py and the rest of Module 2
DAMAGE_CATEGORIES = {
    "screen_display": "Cracked/scratched screen, dead pixels, flickering, no image/blank display, discoloration.",
    "structural_body": "Cracked/broken casing, dents, warping, broken hinge, missing structural piece.",
    "port_connector": "Damaged, bent, or loose port/connector (charging port, USB, HDMI, etc.).",
    "missing_component": "A part, cable, or accessory that should be present is visibly missing.",
    "liquid_moisture": "Visible liquid residue, staining, or corrosion consistent with a spill.",
    "input_control": "Broken/missing key, stuck button, damaged touchpad/touchscreen surface, broken control.",
    "power_boot_failure": "Visible signs consistent with power/battery failure (e.g. swelling, burn marks near battery) - "
                           "NOTE: most power/boot failures have no visible symptom at all and should come from Agent 3, "
                           "not this agent; only use this category if there IS a visible physical sign.",
    "cosmetic_wear": "Minor scuffs, scratches, or wear that does not affect function.",
    "other_unclassified": "Visible damage that doesn't clearly fit any category above, or the image is unclear/ambiguous.",
}

SEVERITY_LEVELS = ["Cosmetic", "Moderate", "Severe", "Critical"]


@dataclass
class DamageClassification:
    damage_type: str = None
    damage_level: str = None
    confidence: float = 0.0
    description: str = ""
    is_new_damage: bool = True
    raw_response: str = ""
    parse_ok: bool = True
    error: str = None


@dataclass
class ConsistencyResult:
    final_damage_type: str = None
    final_damage_level: str = None
    final_description: str = ""
    is_new_damage: bool = True
    agreement_ratio: float = 0.0
    requires_human_review: bool = True
    individual_calls: list = field(default_factory=list)


def _build_prompt(item_type, change_score, alignment_warning=None, resolution_warning=None,

                   has_full_context=False):
    """

    Build the classification prompt, including Agent 1's own diagnostics as

    context - NOT as ground truth, since Agent 1's confidence issues should

    make Agent 2 more cautious, not less.



    has_full_context: True when the full baseline and return photos are

    ALSO provided (not just the crop) - adds an explicit instruction to

    check whether the flagged mark is genuinely NEW (only visible in the

    return photo) or PRE-EXISTING (already visible in the baseline photo,

    e.g. an old scratch the item already had at pickup). Pre-existing

    damage should not be flagged as new damage from this session, even if

    Agent 1's pixel-level diff picked up a region near it.

    """
    category_list = "\n".join(f'  - "{name}": {desc}' for name, desc in DAMAGE_CATEGORIES.items())

    context_notes = []
    if alignment_warning:
        context_notes.append(
            f"NOTE: the automated pre-screening flagged an alignment issue: \"{alignment_warning}\" "
            f"- treat the change_score below as less reliable than usual."
        )
    if resolution_warning:
        context_notes.append(f"NOTE: {resolution_warning}")
    context_block = ("\n".join(context_notes) + "\n") if context_notes else ""

    new_vs_old_instruction = ""
    if has_full_context:
        new_vs_old_instruction = (
            "\nYou have also been given the FULL baseline (pickup) and return photos, not just "
            "the cropped region - use them to check whether the damage shown in the crop is "
            "GENUINELY NEW (visible in the return photo but NOT in the baseline photo) or "
            "PRE-EXISTING (the same mark is already visible in the baseline photo - e.g. an old "
            "scratch or dent the item already had at pickup, which the automated crop detector "
            "may have flagged again due to lighting/angle differences, not because anything "
            "actually changed). If the damage is pre-existing rather than new, set "
            "\"is_new_damage\": false and explain this in your description - pre-existing damage "
            "should not be attributed to this rental session.\n"
        )

    return f"""You are inspecting a cropped photo region of a {item_type} that was automatically \

flagged as visually changed between a pickup photo and a return photo.



Automated pre-screening change_score: {change_score} (0-1 scale, higher = more visual difference \

detected by pixel-level comparison - this is a rough signal, not a judgment of damage type or severity).

{context_block}{new_vs_old_instruction}

Classify the damage shown in this image into EXACTLY ONE of these categories:

{category_list}



Also rate severity as one of: {", ".join(SEVERITY_LEVELS)}.

- Cosmetic: minor, does not affect function.

- Moderate: visible damage, item still usable.

- Severe: damage impairs normal use.

- Critical: item non-functional or unsafe.



If the image is unclear, ambiguous, or doesn't show a clear physical defect, use "other_unclassified" \

and a low confidence score rather than guessing.



Respond with ONLY a JSON object in this exact shape, nothing else - no preamble, no markdown fences:

{{"damage_type": "screen_display", "damage_level": "Severe", "confidence": 0.85, "is_new_damage": true, "description": "A short, human-readable description of what you see, for an admin dashboard."}}

"""


def _read_image_part(image_path):
    with open(image_path, "rb") as f:
        data = f.read()
    mime_type = "image/png" if image_path.lower().endswith(".png") else "image/jpeg"
    return types.Part.from_bytes(data=data, mime_type=mime_type)


def _parse_json_response(text):
    cleaned = text.strip()
    if cleaned.startswith("```"):
        cleaned = cleaned.strip("`")
        if cleaned.lower().startswith("json"):
            cleaned = cleaned[4:]
    return json.loads(cleaned.strip())


def classify_damage_single_call(crop_path, item_type, change_score, client=None,

                                 baseline_path=None, return_path=None,

                                 alignment_warning=None, resolution_warning=None) -> DamageClassification:
    """

    One VLM call classifying a single crop. Optionally includes the full

    baseline/return photos for context - both for cases like a crack near

    a hinge reading differently than one mid-screen, AND (the main reason

    this is now used by default from Agent 2's real entry points) so the

    model can distinguish genuinely NEW damage from PRE-EXISTING damage the

    item already had at pickup.

    """
    if client is None:
        client = genai.Client(api_key=os.environ.get("GOOGLE_API_KEY"))

    has_full_context = baseline_path is not None and return_path is not None
    prompt = _build_prompt(item_type, change_score, alignment_warning, resolution_warning,
                            has_full_context=has_full_context)

    contents = [_read_image_part(crop_path)]
    if baseline_path:
        contents.append("(This is the BASELINE/pickup photo for context:)")
        contents.append(_read_image_part(baseline_path))
    if return_path:
        contents.append("(This is the RETURN photo for context:)")
        contents.append(_read_image_part(return_path))
    contents.append(prompt)

    raw_text = ""
    try:
        response = client.models.generate_content(
            model=VLM_MODEL,
            contents=contents,
            config=types.GenerateContentConfig(temperature=0.2),  # low but non-zero - slight
                                                                    # variation across the self-
                                                                    # consistency calls is useful signal
        )
        raw_text = response.text
        parsed = _parse_json_response(raw_text)

        damage_type = parsed.get("damage_type")
        if damage_type not in DAMAGE_CATEGORIES:
            damage_type = "other_unclassified"

        damage_level = parsed.get("damage_level")
        if damage_level not in SEVERITY_LEVELS:
            damage_level = "Moderate"

        return DamageClassification(
            damage_type=damage_type,
            damage_level=damage_level,
            confidence=float(parsed.get("confidence", 0.5)),
            description=parsed.get("description", ""),
            is_new_damage=bool(parsed.get("is_new_damage", True)),
            raw_response=raw_text,
            parse_ok=True,
        )

    except (json.JSONDecodeError, KeyError, ValueError) as e:
        return DamageClassification(raw_response=raw_text, parse_ok=False,
                                     error=f"Failed to parse VLM response: {e}")
    except Exception as e:
        return DamageClassification(parse_ok=False, error=f"VLM call failed: {e}")


def classify_damage_with_consistency(crop_path, item_type, change_score,

                                      baseline_path=None, return_path=None,

                                      alignment_warning=None, resolution_warning=None,

                                      n_calls=SELF_CONSISTENCY_CALLS,

                                      api_key=None) -> ConsistencyResult:
    """

    Main entry point for Agent 2. Calls the VLM classifier N times on the

    same crop and checks agreement, instead of trusting a single call.



    If the calls agree strongly on damage_type: trust it, requires_human_review=False.

    If they disagree: still return the most common answer as a best guess,

    but flag requires_human_review=True so a human/Agent 4 double-checks

    rather than the system silently acting on an uncertain classification.

    """
    client = genai.Client(api_key=api_key or os.environ.get("GOOGLE_API_KEY"))

    calls = []
    for _ in range(n_calls):
        result = classify_damage_single_call(
            crop_path, item_type, change_score, client=client,
            baseline_path=baseline_path, return_path=return_path,
            alignment_warning=alignment_warning, resolution_warning=resolution_warning,
        )
        calls.append(result)

    valid_calls = [c for c in calls if c.parse_ok]
    if not valid_calls:
        return ConsistencyResult(
            final_damage_type="other_unclassified",
            final_damage_level="Moderate",
            final_description="Classification failed - all VLM calls errored.",
            is_new_damage=True,
            agreement_ratio=0.0,
            requires_human_review=True,
            individual_calls=[c.__dict__ for c in calls],
        )

    type_counts = Counter(c.damage_type for c in valid_calls)
    most_common_type, count = type_counts.most_common(1)[0]
    agreement_ratio = count / len(valid_calls)

    # among calls that agree with the majority, use the one with highest
    # confidence for the final severity/description
    agreeing_calls = [c for c in valid_calls if c.damage_type == most_common_type]
    best_call = max(agreeing_calls, key=lambda c: c.confidence)

    # is_new_damage: trust it only if the calls that agree on damage_type
    # ALSO agree it's new - a mixed signal here should default to treating
    # it as new (safer default - a real new defect should not slip through
    # unflagged just because the model was uncertain about its age)
    new_damage_votes = [c.is_new_damage for c in agreeing_calls]
    is_new_damage = sum(new_damage_votes) >= len(new_damage_votes) / 2

    return ConsistencyResult(
        final_damage_type=most_common_type,
        final_damage_level=best_call.damage_level,
        final_description=best_call.description,
        is_new_damage=is_new_damage,
        agreement_ratio=round(agreement_ratio, 2),
        requires_human_review=(agreement_ratio < AGREEMENT_THRESHOLD),
        individual_calls=[
            {"damage_type": c.damage_type, "damage_level": c.damage_level,
             "confidence": c.confidence, "is_new_damage": c.is_new_damage}
            for c in valid_calls
        ],
    )