Detecting and Editing Visual Objects with Gemini

🛠️ Choice 1 – Gemini API through Vertex AI

Necessities:

Gen AI SDK surroundings variables:

• GOOGLE_GENAI_USE_VERTEXAI="True"
• GOOGLE_CLOUD_PROJECT="<PROJECT_ID>"
• GOOGLE_CLOUD_LOCATION="<LOCATION>"

💡 For preview fashions, the placement have to be set to world. For typically obtainable fashions, we are able to select the closest location among the many Google model endpoint locations.

ℹ️ Be taught extra about setting up a project and a development environment.

🛠️ Choice 2 – Gemini API through Google AI Studio

Requirement:

Gen AI SDK surroundings variables:

• GOOGLE_GENAI_USE_VERTEXAI="False"
• GOOGLE_API_KEY="<API_KEY>"

ℹ️ Be taught extra about getting a Gemini API key from Google AI Studio.

💡 You’ll be able to retailer your surroundings configuration exterior of the supply code:

import os
import sys
from collections.abc import Callable

from google import genai

# Handbook setup (go away unchanged if setup is environment-defined)

# @markdown **Which API: Vertex AI or Google AI Studio?**
GOOGLE_GENAI_USE_VERTEXAI = True  # @param {kind: "boolean"}

# @markdown **Choice A - Google Cloud undertaking [+location]**
GOOGLE_CLOUD_PROJECT = ""  # @param {kind: "string"}
GOOGLE_CLOUD_LOCATION = "world"  # @param {kind: "string"}

# @markdown **Choice B - Google AI Studio API key**
GOOGLE_API_KEY = ""  # @param {kind: "string"}


def check_environment() -> bool:
    check_colab_user_authentication()
    return check_manual_setup() or check_vertex_ai() or check_colab() or check_local()


def check_manual_setup() -> bool:
    return check_define_env_vars(
        GOOGLE_GENAI_USE_VERTEXAI,
        GOOGLE_CLOUD_PROJECT.strip(),  # Might need been pasted with line return
        GOOGLE_CLOUD_LOCATION,
        GOOGLE_API_KEY,
    )


def check_vertex_ai() -> bool:
    # Workbench and Colab Enterprise
    match os.getenv("VERTEX_PRODUCT", ""):
        case "WORKBENCH_INSTANCE":
            go
        case "COLAB_ENTERPRISE":
            if not running_in_colab_env():
                return False
        case _:
            return False

    return check_define_env_vars(
        True,
        os.getenv("GOOGLE_CLOUD_PROJECT", ""),
        os.getenv("GOOGLE_CLOUD_REGION", ""),
        "",
    )


def check_colab() -> bool:
    if not running_in_colab_env():
        return False

    # Colab Enterprise was checked earlier than, so that is Colab solely
    from google.colab import auth as colab_auth  # kind: ignore

    colab_auth.authenticate_user()

    # Use Colab Secrets and techniques (🗝️ icon in left panel) to retailer the surroundings variables
    # Secrets and techniques are personal, seen solely to you and the notebooks that you choose
    # - Vertex AI: Retailer your settings as secrets and techniques
    # - Google AI: Immediately import your Gemini API key from the UI
    vertexai, undertaking, location, api_key = get_vars(get_colab_secret)

    return check_define_env_vars(vertexai, undertaking, location, api_key)


def check_local() -> bool:
    vertexai, undertaking, location, api_key = get_vars(os.getenv)

    return check_define_env_vars(vertexai, undertaking, location, api_key)


def running_in_colab_env() -> bool:
    # Colab or Colab Enterprise
    return "google.colab" in sys.modules


def check_colab_user_authentication() -> None:
    if running_in_colab_env():
        from google.colab import auth as colab_auth  # kind: ignore

        colab_auth.authenticate_user()


def get_colab_secret(secret_name: str, default: str) -> str:
    from google.colab import errors, userdata  # kind: ignore

    attempt:
        return userdata.get(secret_name)
    besides errors.SecretNotFoundError:
        return default


def disable_colab_cell_scrollbar() -> None:
    if running_in_colab_env():
        from google.colab import output  # kind: ignore

        output.no_vertical_scroll()


def get_vars(getenv: Callable[[str, str], str]) -> tuple[bool, str, str, str]:
    # Restrict getenv calls to the minimal (could set off UI affirmation for secret entry)
    vertexai_str = getenv("GOOGLE_GENAI_USE_VERTEXAI", "")
    if vertexai_str:
        vertexai = vertexai_str.decrease() in ["true", "1"]
    else:
        vertexai = bool(getenv("GOOGLE_CLOUD_PROJECT", ""))

    undertaking = getenv("GOOGLE_CLOUD_PROJECT", "") if vertexai else ""
    location = getenv("GOOGLE_CLOUD_LOCATION", "") if undertaking else ""
    api_key = getenv("GOOGLE_API_KEY", "") if not undertaking else ""

    return vertexai, undertaking, location, api_key


def check_define_env_vars(
    vertexai: bool,
    undertaking: str,
    location: str,
    api_key: str,
) -> bool:
    match (vertexai, bool(undertaking), bool(location), bool(api_key)):
        case (True, True, _, _):
            # Vertex AI - Google Cloud undertaking [+location]
            location = location or "world"
            define_env_vars(vertexai, undertaking, location, "")
        case (True, False, _, True):
            # Vertex AI - API key
            define_env_vars(vertexai, "", "", api_key)
        case (False, _, _, True):
            # Google AI Studio - API key
            define_env_vars(vertexai, "", "", api_key)
        case _:
            return False

    return True


def define_env_vars(vertexai: bool, undertaking: str, location: str, api_key: str) -> None:
    os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = str(vertexai)
    os.environ["GOOGLE_CLOUD_PROJECT"] = undertaking
    os.environ["GOOGLE_CLOUD_LOCATION"] = location
    os.environ["GOOGLE_API_KEY"] = api_key


def check_configuration(shopper: genai.Shopper) -> None:
    service = "Vertex AI" if shopper.vertexai else "Google AI Studio"
    print(f"✅ Utilizing the {service} API", finish="")

    if shopper._api_client.undertaking:
        print(f' with undertaking "{shopper._api_client.undertaking[:7]}…"', finish="")
        print(f' in location "{shopper._api_client.location}"')
    elif shopper._api_client.api_key:
        api_key = shopper._api_client.api_key
        print(f' with API key "{api_key[:5]}…{api_key[-5:]}"', finish="")
        print(f" (in case of error, make certain it was created for {service})")


print("✅ Atmosphere capabilities outlined")

from dataclasses import dataclass
from enum import StrEnum

Url = str


class Supply(StrEnum):
    incunable = "https://tile.loc.gov/image-services/iiif/service:rbc:rbc0001:2014:2014rosen0487:0165/full/pct:25/0/default.jpg"
    engravings = "https://tile.loc.gov/image-services/iiif/service:gdc:gdcscd:00:34:07:66:92:1:00340766921:0121/full/pct:50/0/default.jpg"
    museum_guidebook = "https://tile.loc.gov/image-services/iiif/service:rbc:rbc0001:2014:2014gen34181:0033/full/pct:75/0/default.jpg"
    denver_illustrated = "https://tile.loc.gov/image-services/iiif/service:gdc:gdclccn:rc:01:00:04:94:rc01000494:0051/full/pct:50/0/default.jpg"
    physics_textbook = "https://tile.loc.gov/image-services/iiif/service:gdc:gdcscd:00:03:64:87:31:8:00036487318:0103/full/pct:50/0/default.jpg"
    portrait_miniatures = "https://tile.loc.gov/image-services/iiif/service:rbc:rbc0001:2024:2024rosen013592v02:0249/full/pct:50/0/default.jpg"
    wizard_of_oz_drawings = "https://tile.loc.gov/image-services/iiif/service:rbc:rbc0001:2006:2006gen32405:0048/full/pct:25/0/default.jpg"
    work = "https://photographs.unsplash.com/photo-1714146681164-f26fed839692?h=1440"
    alice_drawing = "https://photographs.unsplash.com/photo-1630595011903-689853b04ee2?h=800"
    ebook = "https://photographs.unsplash.com/photo-1643451533573-ee364ba6e330?h=800"
    handbook = "https://photographs.unsplash.com/photo-1623666936367-a100f62ba9b7?h=800"
    electronics = "https://photographs.unsplash.com/photo-1757397584789-8b2c5bfcdbc3?h=1440"


@dataclass
class SourceMetadata:
    title: str
    webpage_url: Url
    credit_line: str


LOC = "Library of Congress"
LOC_RARE_BOOKS = "Library of Congress, Uncommon E-book and Particular Collections Division"
LOC_MEETING_FRONTIERS = "Library of Congress, Assembly of Frontiers"

metadata_by_source: dict[Source, SourceMetadata] = {
    Supply.incunable: SourceMetadata(
        "Vergaderinge der historien van Troy (1485)",
        "https://www.loc.gov/useful resource/rbc0001.2014rosen0487/?sp=165",
        LOC_RARE_BOOKS,
    ),
    Supply.engravings: SourceMetadata(
        "Harper's illustrated catalogue (1847)",
        "https://www.loc.gov/useful resource/gdcscd.00340766921/?sp=121",
        LOC,
    ),
    Supply.museum_guidebook: SourceMetadata(
        "Barnum's American Museum illustrated (1850)",
        "https://www.loc.gov/useful resource/rbc0001.2014gen34181/?sp=33",
        LOC_RARE_BOOKS,
    ),
    Supply.denver_illustrated: SourceMetadata(
        "Denver illustrated (1893)",
        "https://www.loc.gov/useful resource/gdclccn.rc01000494/?sp=51",
        LOC_MEETING_FRONTIERS,
    ),
    Supply.physics_textbook: SourceMetadata(
        "Classes in physics (1916)",
        "https://www.loc.gov/useful resource/gdcscd.00036487318/?sp=103",
        LOC,
    ),
    Supply.portrait_miniatures: SourceMetadata(
        "The historical past of portrait miniatures (1904)",
        "https://www.loc.gov/useful resource/rbc0001.2024rosen013592v02/?sp=249",
        LOC_RARE_BOOKS,
    ),
    Supply.wizard_of_oz_drawings: SourceMetadata(
        "The fantastic Wizard of Oz (1899)",
        "https://www.loc.gov/useful resource/rbc0001.2006gen32405/?sp=48",
        LOC_RARE_BOOKS,
    ),
    Supply.work: SourceMetadata(
        "Open ebook exhibiting work by Vincent van Gogh",
        "https://unsplash.com/pictures/9hD7qrxICag",
        "Photograph by Trung Manh cong on Unsplash",
    ),
    Supply.alice_drawing: SourceMetadata(
        "Open ebook exhibiting an illustration and textual content from Alice's Adventures in Wonderland",
        "https://unsplash.com/pictures/bewzr_Q9u2o",
        "Photograph by Brett Jordan on Unsplash",
    ),
    Supply.ebook: SourceMetadata(
        "Open ebook exhibiting two botanical illustrations",
        "https://unsplash.com/pictures/4IDqcNj827I",
        "Photograph by Ranurte on Unsplash",
    ),
    Supply.handbook: SourceMetadata(
        "Open consumer handbook for classic digital camera",
        "https://unsplash.com/pictures/aaFU96eYASk",
        "Photograph by Annie Spratt on Unsplash",
    ),
    Supply.electronics: SourceMetadata(
        "Circuit board with digital elements",
        "https://unsplash.com/pictures/Aqa1pHQ57pw",
        "Photograph by Albert Stoynov on Unsplash",
    ),
}

print("✅ Check photographs outlined")

from enum import auto
from pathlib import Path
from typing import Any, forged

import IPython.show
import matplotlib.pyplot as plt
import pydantic
import tenacity
from google.genai.errors import ClientError
from google.genai.varieties import (
    FinishReason,
    GenerateContentConfig,
    GenerateContentResponse,
    PIL_Image,
    ThinkingConfig,
    ThinkingLevel,
)


# Multimodal fashions with spatial understanding and structured outputs
class MultimodalModel(StrEnum):
    # Typically Accessible (GA)
    GEMINI_2_5_FLASH = "gemini-2.5-flash"
    GEMINI_2_5_PRO = "gemini-2.5-pro"
    # Preview
    GEMINI_3_FLASH_PREVIEW = "gemini-3-flash-preview"
    GEMINI_3_1_PRO_PREVIEW = "gemini-3.1-pro-preview"
    # Default mannequin used for object detection
    DEFAULT = GEMINI_3_FLASH_PREVIEW


# Picture technology and modifying fashions
class ImageModel(StrEnum):
    # Typically Accessible (GA)
    GEMINI_2_5_FLASH_IMAGE = "gemini-2.5-flash-image"  # Nano Banana 🍌
    # Preview
    GEMINI_3_PRO_IMAGE_PREVIEW = "gemini-3-pro-image-preview"  # Nano Banana Professional 🍌
    # Default mannequin used for picture modifying
    DEFAULT = GEMINI_2_5_FLASH_IMAGE


Mannequin = MultimodalModel | ImageModel


def generate_content(
    contents: record[Any],
    mannequin: Mannequin,
    config: GenerateContentConfig | None,
    should_display_response_info: bool = False,
) -> GenerateContentResponse | None:
    response = None
    shopper = check_client_for_model(mannequin)

    for try in get_retrier():
        with try:
            response = shopper.fashions.generate_content(
                mannequin=mannequin.worth,
                contents=contents,
                config=config,
            )
    if should_display_response_info:
        display_response_info(response, config)

    return response


def check_client_for_model(mannequin: Mannequin) -> genai.Shopper:
    if (
        mannequin.worth.endswith("-preview")
        and shopper.vertexai
        and shopper._api_client.location != "world"
    ):
        # Preview fashions are solely obtainable on the "world" location
        return genai.Shopper(location="world")

    return shopper


def display_response_info(
    response: GenerateContentResponse | None,
    config: GenerateContentConfig | None,
) -> None:
    if response is None:
        print("❌ No response")
        return

    if usage_metadata := response.usage_metadata:
        if usage_metadata.prompt_token_count:
            print(f"Enter tokens   : {usage_metadata.prompt_token_count:9,d}")
        if usage_metadata.candidates_token_count:
            print(f"Output tokens  : {usage_metadata.candidates_token_count:9,d}")
        if usage_metadata.thoughts_token_count:
            print(f"Ideas tokens: {usage_metadata.thoughts_token_count:9,d}")

    if (
        config shouldn't be None
        and config.response_mime_type == "software/json"
        and response.parsed is None
    ):
        print("❌ Couldn't parse the JSON response")
        return
    if not response.candidates:
        print("❌ No `response.candidates`")
        return
    if (finish_reason := response.candidates[0].finish_reason) != FinishReason.STOP:
        print(f"❌ {finish_reason = }")
    if not response.textual content:
        print("❌ No `response.textual content`")
        return


def generate_image(
    sources: record[PIL_Image],
    immediate: str,
    mannequin: ImageModel,
    config: GenerateContentConfig | None = None,
) -> PIL_Image | None:
    contents = [*sources, prompt.strip()]

    response = generate_content(contents, mannequin, config)

    return check_get_output_image_from_response(response)


def check_get_output_image_from_response(
    response: GenerateContentResponse | None,
) -> PIL_Image | None:
    if response is None:
        print("❌ No `response`")
        return None
    if not response.candidates:
        print("❌ No `response.candidates`")
        if response.prompt_feedback:
            if block_reason := response.prompt_feedback.block_reason:
                print(f"{block_reason = :s}")
            if block_reason_message := response.prompt_feedback.block_reason_message:
                print(f"{block_reason_message = }")
        return None
    if not (content material := response.candidates[0].content material):
        print("❌ No `response.candidates[0].content material`")
        return None
    if not (components := content material.components):
        print("❌ No `response.candidates[0].content material.components`")
        return None

    output_image: PIL_Image | None = None
    for half in components:
        if half.textual content:
            display_markdown(half.textual content)
            proceed
        sdk_image = half.as_image()
        assert sdk_image shouldn't be None
        output_image = sdk_image._pil_image
        assert output_image shouldn't be None
        break  # There must be a single picture

    return output_image


def get_thinking_config(mannequin: Mannequin) -> ThinkingConfig | None:
    match mannequin:
        case MultimodalModel.GEMINI_2_5_FLASH:
            return ThinkingConfig(thinking_budget=0)
        case MultimodalModel.GEMINI_2_5_PRO:
            return ThinkingConfig(thinking_budget=128, include_thoughts=False)
        case MultimodalModel.GEMINI_3_FLASH_PREVIEW:
            return ThinkingConfig(thinking_level=ThinkingLevel.MINIMAL)
        case MultimodalModel.GEMINI_3_1_PRO_PREVIEW:
            return ThinkingConfig(thinking_level=ThinkingLevel.LOW)
        case _:
            return None  # Default


def display_markdown(markdown: str) -> None:
    IPython.show.show(IPython.show.Markdown(markdown))


def display_image(picture: PIL_Image) -> None:
    IPython.show.show(picture)


def get_retrier() -> tenacity.Retrying:
    return tenacity.Retrying(
        cease=tenacity.stop_after_attempt(7),
        wait=tenacity.wait_incrementing(begin=10, increment=1),
        retry=should_retry_request,
        reraise=True,
    )


def should_retry_request(retry_state: tenacity.RetryCallState) -> bool:
    if not retry_state.consequence:
        return False
    err = retry_state.consequence.exception()
    if not isinstance(err, ClientError):
        return False
    print(f"❌ ClientError {err.code}: {err.message}")

    retry = False
    match err.code:
        case 400 if err.message shouldn't be None and " attempt once more " in err.message:
            # Workshop: first time entry to Cloud Storage (service agent provisioning)
            retry = True
        case 429:
            # Workshop: non permanent undertaking with 1 QPM quota
            retry = True
    print(f"🔄 Retry: {retry}")

    return retry


print("✅ Helpers outlined")

import io
import urllib.request
from collections.abc import Iterator
from dataclasses import subject
from datetime import datetime

import PIL.Picture
from google.genai.varieties import Half, PartMediaResolutionLevel
from PIL.PngImagePlugin import PngInfo

OBJECT_DETECTION_PROMPT = """
Detect each illustration inside the ebook photograph and extract the next information for every:
- `box_2d`: Bounding field coordinates of the illustration solely (ignoring any caption).
- `caption`: Verbatim caption or legend reminiscent of "Determine 1". Use "" if not discovered.
- `label`: Single-word label describing the illustration. Use "" if not discovered.
"""

# Margin added to detected/cropped objects, giving extra context for a greater understanding of spatial distortions
CROP_MARGIN_PX = 10

# Set to True to avoid wasting every generated picture
SAVE_GENERATED_IMAGES = False
OUTPUT_IMAGES_PATH = Path("./object_detection_and_editing")


# Matching class for structured output technology
class DetectedObject(pydantic.BaseModel):
    box_2d: record[int]
    caption: str
    label: str


# Misc information courses
InputImage = Path | Url
DetectedObjects = record[DetectedObject]
WorkflowStepImages = record[PIL_Image]


class WorkflowStep(StrEnum):
    SOURCE = auto()
    CROPPED = auto()
    RESTORED = auto()
    COLORIZED = auto()
    CINEMATIZED = auto()


@dataclass
class VisualObjectWorkflow:
    source_image: PIL_Image
    detected_objects: DetectedObjects
    images_by_step: dict[WorkflowStep, WorkflowStepImages] = subject(default_factory=dict)

    def __post_init__(self) -> None:
        denormalize_bounding_boxes(self)


workflow_by_image: dict[InputImage, VisualObjectWorkflow] = {}


def denormalize_bounding_boxes(self: VisualObjectWorkflow) -> None:
    """Convert the box_2d coordinates.
    - Earlier than: [y1, x1, y2, x2] normalized to 0-1000, as returned by Gemini
    - After:  [x1, y1, x2, y2] in source_image coordinates, as utilized in Pillow
    """

    def to_image_coord(coord: int, dim: int) -> int:
        return int(coord * dim / 1000 + 0.5)

    w, h = self.source_image.dimension
    for obj in self.detected_objects:
        y1, x1, y2, x2 = obj.box_2d
        x1, x2 = to_image_coord(x1, w), to_image_coord(x2, w)
        y1, y2 = to_image_coord(y1, h), to_image_coord(y2, h)
        obj.box_2d = [x1, y1, x2, y2]


def detect_objects(
    picture: InputImage,
    immediate: str = OBJECT_DETECTION_PROMPT,
    mannequin: MultimodalModel = MultimodalModel.DEFAULT,
    config: GenerateContentConfig | None = None,
    media_resolution: PartMediaResolutionLevel | None = None,
    display_results: bool = True,
) -> None:
    display_image_source_info(picture)
    pil_image, content_part = get_pil_image_and_part(picture, mannequin, media_resolution)
    immediate = immediate.strip()
    contents = [content_part, prompt]
    config = config or get_object_detection_config(mannequin)

    response = generate_content(contents, mannequin, config)

    if response shouldn't be None and response.parsed shouldn't be None:
        detected_objects = forged(DetectedObjects, response.parsed)
    else:
        detected_objects = DetectedObjects()

    workflow = VisualObjectWorkflow(pil_image, detected_objects)
    workflow_by_image[image] = workflow
    add_cropped_objects(workflow, picture, immediate)

    if display_results:
        display_detected_objects(workflow)


def get_pil_image_and_part(
    picture: InputImage,
    mannequin: MultimodalModel,
    media_resolution: PartMediaResolutionLevel | None,
) -> tuple[PIL_Image, Part]:
    if isinstance(picture, Path):
        image_bytes = picture.read_bytes()
    else:
        headers = {"Person-Agent": "Mozilla/5.0"}
        req = urllib.request.Request(picture, headers=headers)
        with urllib.request.urlopen(req, timeout=10) as response:
            image_bytes = response.learn()

    pil_image = PIL.Picture.open(io.BytesIO(image_bytes))
    content_part = Half.from_bytes(
        information=image_bytes,
        mime_type="picture/*",
        media_resolution=media_resolution,
    )

    return pil_image, content_part


def get_object_detection_config(mannequin: Mannequin) -> GenerateContentConfig:
    # Low randomness for extra determinism
    return GenerateContentConfig(
        temperature=0.0,
        top_p=0.0,
        seed=42,
        response_mime_type="software/json",
        response_schema=DetectedObjects,
        thinking_config=get_thinking_config(mannequin),
    )


def add_cropped_objects(
    workflow: VisualObjectWorkflow,
    enter: InputImage,
    immediate: str,
    crop_margin: int = CROP_MARGIN_PX,
) -> None:
    cropped_images: record[PIL_Image] = []
    obj_count = len(workflow.detected_objects)
    for obj_order, obj in enumerate(workflow.detected_objects, 1):
        cropped_image, _ = extract_object_image(workflow.source_image, obj, crop_margin)
        cropped_images.append(cropped_image)
        save_workflow_image(
            WorkflowStep.SOURCE,
            WorkflowStep.CROPPED,
            enter,
            obj_order,
            obj_count,
            cropped_image,
            dict(immediate=immediate, crop_margin=str(crop_margin)),
        )
    workflow.images_by_step[WorkflowStep.CROPPED] = cropped_images


def extract_object_image(
    picture: PIL_Image,
    obj: DetectedObject,
    margin: int = 0,
) -> tuple[PIL_Image, tuple[int, int, int, int]]:
    def clamp(coord: int, dim: int) -> int:
        return min(max(coord, 0), dim)

    x1, y1, x2, y2 = obj.box_2d
    w, h = picture.dimension
    if margin != 0:
        x1, x2 = clamp(x1 - margin, w), clamp(x2 + margin, w)
        y1, y2 = clamp(y1 - margin, h), clamp(y2 + margin, h)

    field = (x1, y1, x2, y2)
    object_image = picture.crop(field)

    return object_image, field


def save_workflow_image(
    source_step: WorkflowStep,
    target_step: WorkflowStep,
    input_image: InputImage,
    obj_order: int,
    obj_count: int,
    target_image: PIL_Image | None,
    image_info: dict[str, str] | None = None,
) -> None:
    if not SAVE_GENERATED_IMAGES or target_image is None:
        return
    if not OUTPUT_IMAGES_PATH.is_dir():
        OUTPUT_IMAGES_PATH.mkdir(dad and mom=True)
    time_str = datetime.now().strftime("%Y-%m-%d_percentH-%M-%S")
    attempt:
        filename = f"{Supply(input_image).title}_"
    besides ValueError:
        filename = ""
    filename += f"{obj_order}o{obj_count}_{source_step}_{target_step}_{time_str}.png"
    image_path = OUTPUT_IMAGES_PATH.joinpath(filename)
    params = {}
    if image_info:
        png_info = PngInfo()
        for okay, v in image_info.objects():
            png_info.add_text(okay, v)
        params.replace(pnginfo=png_info)
    target_image.save(image_path, **params)


# Matplotlib
FIGURE_FG_COLOR = "#F1F3F4"
FIGURE_BG_COLOR = "#202124"
EDGE_COLOR = "#80868B"
rcParams = {
    "determine.dpi": 300,
    "textual content.colour": FIGURE_FG_COLOR,
    "determine.edgecolor": FIGURE_FG_COLOR,
    "axes.titlecolor": FIGURE_FG_COLOR,
    "axes.edgecolor": FIGURE_FG_COLOR,
    "xtick.colour": FIGURE_FG_COLOR,
    "ytick.colour": FIGURE_FG_COLOR,
    "determine.facecolor": FIGURE_BG_COLOR,
    "axes.edgecolor": EDGE_COLOR,
    "xtick.backside": False,
    "xtick.high": False,
    "ytick.left": False,
    "ytick.proper": False,
    "xtick.labelbottom": False,
    "ytick.labelleft": False,
}
plt.rcParams.replace(rcParams)


def display_image_source_info(picture: InputImage) -> None:
    def get_image_info_md() -> str:
        if picture not in Supply:
            return f"[[Source Image]({picture})]"
        supply = Supply(picture)
        metadata = metadata_by_source.get(supply)
        if not metadata:
            return f"[[Source Image]({supply.worth})]"
        components = [
            f"[Source Image]({supply.worth})",
            f"[Source Page]({metadata.webpage_url})",
            metadata.title,
            metadata.credit_line,
        ]
        separator = "•"
        inner_info = f" {separator} ".be part of(components)
        return f"{separator} {inner_info} {separator}"

    def yield_md_rows() -> Iterator[str]:
        horizontal_line = "---"
        image_info = get_image_info_md()
        yield horizontal_line
        yield f"_{image_info}_"
        yield horizontal_line

    display_markdown(f"{chr(10)}{chr(10)}".be part of(yield_md_rows()))


def display_detected_objects(workflow: VisualObjectWorkflow) -> None:
    source_image = workflow.source_image
    detected_objects = PIL.Picture.new("RGB", source_image.dimension, "white")
    for obj in workflow.detected_objects:
        obj_image, field = extract_object_image(source_image, obj)
        detected_objects.paste(obj_image, (field[0], field[1]))

    _, (ax1, ax2) = plt.subplots(1, 2, format="compressed")
    ax1.imshow(source_image)
    ax2.imshow(detected_objects)

    disable_colab_cell_scrollbar()
    plt.present()


print("✅ Object detection helpers outlined")

from collections import defaultdict


def display_detection_data(supply: Supply, show_consolidated: bool = False) -> None:
    def string_with_visible_linebreaks(s: str) -> str:
        return f'''"{s.exchange(chr(10), "↩️")}"'''

    def yield_md_rows_consolidated(workflow: VisualObjectWorkflow) -> Iterator[str]:
        yield "| label | rely | captions |"
        yield "| :--- | ---: | :--- |"
        stats = defaultdict(record)
        for obj in workflow.detected_objects:
            stats[obj.label].append(string_with_visible_linebreaks(obj.caption))
        for label, captions in stats.objects():
            rely = len(captions)
            label_captions = " • ".be part of(sorted(captions))
            yield f"| {label} | {rely} | {label_captions} |"

    def yield_md_rows_with_bbox(workflow: VisualObjectWorkflow) -> Iterator[str]:
        yield "| box_2d | label | caption |"
        yield "| :--- | :--- | :--- |"
        for obj in workflow.detected_objects:
            yield f"| {obj.box_2d} | {obj.label} | {string_with_visible_linebreaks(obj.caption)} |"

    workflow = workflow_by_image.get(supply)
    if workflow is None:
        print(f'❌ No detection for supply "{supply.title}"')
        return
    md_rows = record(
        yield_md_rows_consolidated(workflow)
        if show_consolidated
        else yield_md_rows_with_bbox(workflow)
    )
    display_image_source_info(supply)
    display_markdown(chr(10).be part of(md_rows))

from typing import Protocol


class ObjectEditingFunction(Protocol):
    def __call__(
        self,
        picture: InputImage,
        immediate: str | None = None,
        mannequin: ImageModel | None = None,
        config: GenerateContentConfig | None = None,
        display_results: bool = True,
    ) -> None: ...


SourceTargetSteps = tuple[WorkflowStep, WorkflowStep]
registered_functions: dict[SourceTargetSteps, ObjectEditingFunction] = {}

DEFAULT_EDITING_CONFIG = GenerateContentConfig(response_modalities=["IMAGE"])
EMPTY_IMAGE = PIL.Picture.new("1", (1, 1), "white")


def object_editing_function(
    default_prompt: str,
    source_step: WorkflowStep,
    target_step: WorkflowStep,
    default_model: ImageModel = ImageModel.DEFAULT,
    default_config: GenerateContentConfig = DEFAULT_EDITING_CONFIG,
) -> ObjectEditingFunction:
    def editing_function(
        picture: InputImage,
        immediate: str | None = default_prompt,
        mannequin: ImageModel | None = default_model,
        config: GenerateContentConfig | None = default_config,
        display_results: bool = True,
    ) -> None:
        workflow, source_images = get_workflow_and_step_images(picture, source_step)
        if immediate is None:
            immediate = default_prompt
        immediate = immediate.strip()
        if mannequin is None:
            mannequin = default_model
        # Be aware: "config is None" is legitimate and can use the mannequin endpoint default config

        target_images: record[PIL_Image] = []
        display_image_source_info(picture)
        obj_count = len(source_images)
        for obj_order, source_image in enumerate(source_images, 1):
            target_image = generate_image([source_image], immediate, mannequin, config)
            save_workflow_image(
                source_step,
                target_step,
                picture,
                obj_order,
                obj_count,
                target_image,
                dict(immediate=immediate),
            )
            target_images.append(target_image if target_image else EMPTY_IMAGE)

        workflow.images_by_step[target_step] = target_images
        if display_results:
            display_sources_and_targets(workflow, source_step, target_step)

    registered_functions[(source_step, target_step)] = editing_function

    return editing_function


def get_workflow_and_step_images(
    picture: InputImage,
    step: WorkflowStep,
) -> tuple[VisualObjectWorkflow, list[PIL_Image]]:
    # Objects detected?
    if picture not in workflow_by_image:
        detect_objects(picture, display_results=False)
    workflow = workflow_by_image.get(picture)
    assert workflow shouldn't be None

    # Workflow step objects? (single degree, could possibly be prolonged to a dynamical graph)
    operation = (WorkflowStep.CROPPED, step)
    if step not in workflow.images_by_step and operation in registered_functions:
        source_function = registered_functions[operation]
        source_function(picture, display_results=False)

    # Supply photographs
    source_images = workflow.images_by_step.get(step)
    assert source_images shouldn't be None

    return workflow, source_images


def display_sources_and_targets(
    workflow: VisualObjectWorkflow,
    source_step: WorkflowStep,
    target_step: WorkflowStep,
) -> None:
    source_images = workflow.images_by_step[source_step]
    target_images = workflow.images_by_step[target_step]
    if not source_images:
        print("❌ No photographs to show")
        return

    fig = plt.determine(format="compressed")
    if horizontal := (len(source_images) >= 2):
        rows, cols = 2, len(source_images)
    else:
        rows, cols = len(source_images), 2
    gs = fig.add_gridspec(rows, cols)

    for i, (source_image, target_image) in enumerate(
        zip(source_images, target_images, strict=True)
    ):
        for dim, picture in enumerate([source_image, target_image]):
            grid_spec = gs[dim, i] if horizontal else gs[i, dim]
            ax = fig.add_subplot(grid_spec)
            ax.set_axis_off()
            ax.imshow(picture)

    disable_colab_cell_scrollbar()
    plt.present()


print("✅ Object modifying helpers outlined")