Update main.py

main.py

@@ -9,13 +9,13 @@ with app.setup:
     import ast, base64, glob, io, json, mimetypes, os, re, tempfile, time, zipfile
     from typing import Any, Dict, List, Optional, Union, Callable, Literal
     from pathlib import Path
-    
+
     # --- Third Party Libraries
     from dotenv import load_dotenv
     from ibm_watsonx_ai import APIClient, Credentials
     from ibm_watsonx_ai.foundation_models import ModelInference
-    from kafka import KafkaProducer
-    
+    from kafka import KafkaProducer, KafkaAdminClient
+
     load_dotenv()
     from PIL import Image
     import marimo as mo
@@ -43,25 +43,24 @@ def _():
         process_multiple_images_with_display_data,
         process_multiple_images_with_examples,
     )
-    from samples.image_example_message import image_example_message as example_message
+    from samples.image_example_message import (
+        image_example_message as example_message,
+    )
+
     return (
-        create_data_url,
-        convert_heic_to_jpg,
         create_multiple_image_previews_with_conversion,
         display_results_stack_with_data,
-        process_multiple_images_with_display_data,
-        process_multiple_images_with_examples,
         example_message,
-        wx_regions,
+        process_multiple_images_with_examples,
     )
 
 
 @app.cell
-def _(os):
+def _():
     user = os.environ.get("KAFKA_USER") or ""
     password = os.environ.get("KAFKA_PASSWORD") or ""
     kafka_bootstrap_servers = os.environ.get("KAFKA_BOOTSTRAP_SERVERS") or ""
-    kafka_topic = os.environ.get("KAFKA_TOPIC") or ""
+    kafka_topic_filter = os.environ.get("KAFKA_TOPIC_PREFIX") or ""
     prompt_template = os.environ.get("EXTRACTION_PROMPT") or ""
     wx_creds = {
         "api_key": os.environ.get("WX_APIKEY") or "",
@@ -72,16 +71,17 @@ def _(os):
         "url": os.environ.get("WX_URL") or "https://us-south.ml.cloud.ibm.com",
     }
     return (
-        wx_creds,
-        prompt_template,
-        kafka_bootstrap_servers,
         kafka_bootstrap_servers,
+        kafka_topic_filter,
         password,
-        user
+        prompt_template,
+        user,
+        wx_creds,
     )
 
+
 @app.cell
-def _(certifi, kafka_bootstrap_servers, os, password, user, wx_creds):
+def _(kafka_bootstrap_servers, password, user, wx_creds):
     kafka_config = {
         "bootstrap_servers": kafka_bootstrap_servers.split(","),
         "security_protocol": "SASL_SSL",
@@ -107,20 +107,24 @@ def _(certifi, kafka_bootstrap_servers, os, password, user, wx_creds):
     chat_model_id = wx_creds["model_id"] or "mistralai/mistral-medium-2505"
     return chat_model_id, kafka_config, params
 
+
 @app.cell
-def _(APIClient, Credentials, wx_creds):
+def _(wx_creds):
     wx_credentials = Credentials(url=wx_creds["url"], api_key=wx_creds["api_key"])
     client = (
         APIClient(credentials=wx_credentials, project_id=wx_creds["project_id"])
         if wx_creds["project_id"]
-        else APIClient(credentials=wx_credentials, space_id=wx_creds["space_id"])
-        if wx_creds["space_id"]
-        else APIClient(credentials=wx_credentials)
+        else (
+            APIClient(credentials=wx_credentials, space_id=wx_creds["space_id"])
+            if wx_creds["space_id"]
+            else APIClient(credentials=wx_credentials)
+        )
     )
     return (client,)
 
+
 @app.cell
-def _(ModelInference, chat_model_id, client, params):
+def _(chat_model_id, client, params):
     chat_model = ModelInference(
         api_client=client, model_id=chat_model_id, params=params
     )
@@ -128,7 +132,50 @@ def _(ModelInference, chat_model_id, client, params):
 
 
 @app.cell
-def _(kafka_config, KafkaProducer):
+def _(kafka_config):
+    kafka_admin = KafkaAdminClient(**kafka_config)
+    kafka_topics = kafka_admin.describe_topics()
+    return (kafka_topics,)
+
+
+@app.cell
+def _(kafka_topic_filter, kafka_topics):
+    topic_names = (
+        get_topic_names(kafka_topics, kafka_topic_filter)
+        if kafka_topics
+        else ["placeholder_topic"]
+    )
+    return (topic_names,)
+
+
+@app.cell
+def _(topic_names):
+    kafka_topic_selector = mo.ui.dropdown(
+        topic_names,
+        label="**Select the Target Topic:**",
+        searchable=True,
+        allow_select_none=False,
+        value=topic_names[0],
+    )
+    return (kafka_topic_selector,)
+
+
+@app.cell
+def _(kafka_topic_selector):
+    kafka_topic = kafka_topic_selector.value
+    return (kafka_topic,)
+
+
+@app.function
+def get_topic_names(kafka_topics, filter_word=None):
+    topics = [topic["topic"] for topic in kafka_topics]
+    if filter_word:
+        topics = [t for t in topics if filter_word in t]
+    return topics
+
+
+@app.cell
+def _(kafka_config):
     kafka_producer = KafkaProducer(
         **kafka_config,
         value_serializer=lambda x: x.encode("utf-8") if isinstance(x, str) else x,
@@ -137,7 +184,7 @@ def _(kafka_config, KafkaProducer):
 
 
 @app.cell
-def _(mo):
+def _():
     image_upload = mo.ui.file(
         kind="area",
         filetypes=[".png", ".jpg", ".jpeg", ".tiff", ".heic"],
@@ -148,7 +195,7 @@ def _(mo):
 
 
 @app.cell
-def _(image_upload, mo):
+def _(image_upload):
     if image_upload.name():
         name_printout = mo.md(f"**{image_upload.name()}**")
     else:
@@ -157,23 +204,26 @@ def _(image_upload, mo):
     image_uploader = mo.vstack(
         [image_upload, name_printout], justify="space-around", align="center"
     )
-    return
+    return (image_uploader,)
 
 
 @app.cell
-def _(mo, prompt_template):
-    prompt_editor = mo.md("""
+def _(prompt_template):
+    prompt_editor = mo.md(
+        """
         #### **Provide your instruction prompt here by editing the template:**
 
         {editor}
 
-        """).batch(
+        """
+    ).batch(
         editor=mo.ui.code_editor(
             value=prompt_template, language="markdown", min_height=200
         )
     )
     return (prompt_editor,)
 
+
 @app.function
 def check_state(variable, if_present=False, if_not_present=True):
     return if_not_present if not variable else if_present
@@ -188,12 +238,13 @@ def _(image_upload):
 
 
 @app.cell
-def _(button_disabled, mo):
+def _(button_disabled):
     extract_text_button = mo.ui.run_button(
         label="Extract Text from Images", disabled=button_disabled
     )
     return (extract_text_button,)
 
+
 @app.cell
 def _(prompt_editor, prompt_template):
     instruction_prompt = (
@@ -201,23 +252,33 @@ def _(prompt_editor, prompt_template):
     )
     return (instruction_prompt,)
 
+
 @app.cell
-def _(mo, image_uploader, extract_text_button):
+def _():
+    title = mo.md(
+        """### **GhostEyes:** watsonx.ai Based Image to Mural Sticky Note Converter"""
+    )
+    return (title,)
+
+
+@app.cell
+def _(extract_text_button, image_uploader, kafka_topic_selector, title):
     mo.vstack(
-        [image_uploader, extract_text_button],
+        [title, image_uploader, kafka_topic_selector, extract_text_button],
         align="center",
         gap=2,
     )
     return
 
+
 @app.cell
-def _(mo, multiple_image_previews, results_df):
+def _(multiple_image_previews, results_df):
     extract_stack = mo.vstack(
         [multiple_image_previews, results_df],
         align="center",
         gap=2,
     )
-    return
+    return (extract_stack,)
 
 
 @app.cell
@@ -227,6 +288,7 @@ def _(create_multiple_image_previews_with_conversion, image_upload):
     )
     return (multiple_image_previews,)
 
+
 @app.cell
 def _(
     chat_model,
@@ -253,9 +315,9 @@ def _(
     else:
         results_df = display_files = None
         results_ready = False
-
     return display_files, results_df, results_ready
 
+
 @app.cell
 def _(display_files, display_results_stack_with_data, results_df):
     review_stack = (
@@ -263,16 +325,11 @@ def _(display_files, display_results_stack_with_data, results_df):
         if results_df is not None
         else None
     )
-    return
+    return (review_stack,)
+
 
 @app.cell
-def _(
-    kafka_producer,
-    kafka_topic,
-    results_df,
-    results_ready,
-    send_results_to_kafka,
-):
+def _(kafka_producer, kafka_topic, results_df, results_ready):
     send_kafka_events = (
         send_results_to_kafka(
             kafka_producer, kafka_topic, results_df, column_to_send="model_response"
@@ -282,13 +339,12 @@ def _(
     )
     return
 
+
 @app.cell
-def _(pillow_heif):
+def _():
     pillow_heif.register_heif_opener()
     return
 
-# --- --- --- --- --- --- ---
-
 
 @app.cell
 def _(extract_stack):
@@ -299,245 +355,38 @@ def _(extract_stack):
     return
 
 
-
 @app.cell
 def _(review_stack):
-    ui_accordion_section_2 = mo.accordion(
-        {"**Review Outputs Tab**": review_stack}
-    )
+    ui_accordion_section_2 = mo.accordion({"**Review Outputs Tab**": review_stack})
     ui_accordion_section_2
     return
 
 
-@app.cell
-def _(time):
-    def send_results_to_kafka(
-        kafka_producer,
-        kafka_topic,
-        results_df,
-        exclude_value="No Text Detected",
-        column_to_send="model_response",
-        sleep_time=0.2,
-    ):
-        """
-        Send DataFrame results to Kafka topic, excluding specified values.
-
-        Args:
-            kafka_producer: Kafka producer instance
-            kafka_topic: Kafka topic name
-            results_df: DataFrame containing results
-            exclude_value: Value to exclude from sending (default: "No Text Detected")
-            column_to_send: Column name to send (default: "model_response")
-            sleep_time: Time to sleep between sends in seconds (default: 0.2)
-        """
-        for _, row in results_df.iterrows():
-            value = row[column_to_send]
-            if value != exclude_value:
-                kafka_producer.send(topic=kafka_topic, value=str(value))
-                time.sleep(sleep_time)
-    return (send_results_to_kafka,)
-
-# --- --- --- --- --- --- --- In case you want to use the standard watsonx.ai client instantiation form i use in other notebooks
-
-# @app.cell
-# def _(wx_creds, wx_regions):
-#     client_instantiation_form = (
-#         mo.md(
-#             """
-#         ###**watsonx.ai credentials:**
-
-#         {wx_region}
-
-#         {wx_api_key}
-
-#         {project_id}
-
-#         {space_id}
-#     """
-#         )
-#         .batch(
-#             wx_region=mo.ui.dropdown(
-#                 wx_regions,
-#                 label="Select your watsonx.ai region:",
-#                 value=wx_creds["region"],
-#                 searchable=True,
-#             ),
-#             wx_api_key=mo.ui.text(
-#                 placeholder="Add your IBM Cloud api-key...",
-#                 label="IBM Cloud Api-key:",
-#                 kind="password",
-#                 value=wx_creds["api_key"],
-#             ),
-#             project_id=mo.ui.text(
-#                 placeholder="Add your watsonx.ai project_id...",
-#                 label="Project_ID:",
-#                 kind="text",
-#                 value=wx_creds["project_id"],
-#             ),
-#             space_id=mo.ui.text(
-#                 placeholder="Add your watsonx.ai space_id...",
-#                 label="Space_ID:",
-#                 kind="text",
-#                 value=wx_creds["space_id"],
-#             ),
-#         )
-#         .form(show_clear_button=True, bordered=False)
-#     )
-#     return (client_instantiation_form,)
-
-# @app.cell
-# def _(client_instantiation_form):
-#     client_setup = client_instantiation_form.value or None
-
-#     ### Extract Credential Variables:
-#     if client_setup:
-#         wx_url = client_setup["wx_region"] if client_setup["wx_region"] else "EU"
-#         wx_api_key = (
-#             client_setup["wx_api_key"].strip()
-#             if client_setup["wx_api_key"]
-#             else None
-#         )
-#         os.environ["WATSONX_APIKEY"] = wx_api_key or ""
-
-#         project_id = (
-#             client_setup["project_id"].strip()
-#             if client_setup["project_id"]
-#             else None
-#         )
-#         space_id = (
-#             client_setup["space_id"].strip() if client_setup["space_id"] else None
-#         )
-#     else:
-#         os.environ["WATSONX_APIKEY"] = ""
-#         project_id = space_id = wx_api_key = wx_url = None
-#     return client_setup, project_id, space_id, wx_api_key, wx_url
-
-# @app.cell
-# def _(client_setup, project_id, space_id, wx_api_key, wx_url):
-#     ### Instantiate the watsonx.ai client
-#     if client_setup:
-#         try:
-#             wx_credentials = Credentials(url=wx_url, api_key=wx_api_key)
-#             project_client = (
-#                 APIClient(credentials=wx_credentials, project_id=project_id)
-#                 if project_id
-#                 else None
-#             )
-#             deployment_client = (
-#                 APIClient(credentials=wx_credentials, space_id=space_id)
-#                 if space_id
-#                 else None
-#             )
-#             instantiation_success = True
-#             instantiation_error = None
-#         except Exception as e:
-#             instantiation_success = False
-#             instantiation_error = str(e)
-#             wx_credentials = project_client = deployment_client = None
-#     else:
-#         wx_credentials = project_client = deployment_client = None
-#         instantiation_success = None
-#         instantiation_error = None
-#     return (
-#         deployment_client,
-#         instantiation_error,
-#         instantiation_success,
-#         project_client,
-#     )
-
-
-# @app.cell
-# def _(client_callout_kind, client_instantiation_form, client_status):
-#     client_callout = mo.callout(client_status, kind=client_callout_kind)
-
-#     client_section = mo.hstack(
-#         [client_instantiation_form, client_callout],
-#         align="center",
-#         justify="space-around",
-#     )
-#     return (client_section,)
-
-# @app.cell
-# def _(
-#     client_key,
-#     client_options,
-#     client_selector,
-#     client_setup,
-#     get_key_by_value,
-#     instantiation_error,
-#     instantiation_success,
-#     wrap_with_spaces,
-# ):
-#     active_client_name = (
-#         get_key_by_value(client_options, client_key)
-#         if client_key
-#         else "No Client" or "Project Client"
-#     )
-
-#     if client_setup:
-#         if instantiation_success:
-#             client_status = mo.md(
-#                 f"### ✅ Client Instantiation Successful ✅\n\n"
-#                 f"{client_selector}\n\n"
-#                 f"**Active Client:**{wrap_with_spaces(active_client_name, prefix_spaces=5)}"
-#             )
-#             client_callout_kind = "success"
-#         else:
-#             client_status = mo.md(
-#                 f"### ❌ Client Instantiation Failed\n**Error:** {instantiation_error}\n\nCheck your region selection and credentials"
-#             )
-#             client_callout_kind = "danger"
-#     else:
-#         client_status = mo.md(
-#             f"### Client Instantiation Status will turn Green When Ready\n\n"
-#             f"{client_selector}\n\n"
-#             f"**Active Client:**{wrap_with_spaces(active_client_name, prefix_spaces=5)}"
-#         )
-#         client_callout_kind = "neutral"
-#     return client_callout_kind, client_status
-
-
-# @app.cell
-# def _(deployment_client, project_client):
-#     if project_client is not None and deployment_client is not None:
-#         client_options = {
-#             "Project Client": project_client,
-#             "Deployment Client": deployment_client,
-#         }
-
-#     elif project_client is not None:
-#         client_options = {"Project Client": project_client}
-
-#     elif deployment_client is not None:
-#         client_options = {"Deployment Client": deployment_client}
-
-#     else:
-#         client_options = {"No Client": "Instantiate a Client"}
-
-#     default_client = next(iter(client_options))
-#     client_selector = mo.ui.dropdown(
-#         client_options, value=default_client, label="**Switch your active client:**"
-#     )
-#     return client_options, client_selector
-
-
-# @app.cell
-# def _(client_selector):
-#     client_key = client_selector.value
-#     if client_key == "Instantiate a Client":
-#         client = None
-#     else:
-#         client = client_key
-#     return client, client_key
-
-
-# @app.cell
-# def _(client_section):
-#     ui_accordion_section_1 = mo.accordion(
-#         {"Section 1: **watsonx.ai Credentials**": client_section}
-#     )
-#     ui_accordion_section_1
-#     return
+@app.function
+def send_results_to_kafka(
+    kafka_producer,
+    kafka_topic,
+    results_df,
+    exclude_value="No Text Detected",
+    column_to_send="model_response",
+    sleep_time=0.2,
+):
+    """
+    Send DataFrame results to Kafka topic, excluding specified values.
+
+    Args:
+        kafka_producer: Kafka producer instance
+        kafka_topic: Kafka topic name
+        results_df: DataFrame containing results
+        exclude_value: Value to exclude from sending (default: "No Text Detected")
+        column_to_send: Column name to send (default: "model_response")
+        sleep_time: Time to sleep between sends in seconds (default: 0.2)
+    """
+    for _, row in results_df.iterrows():
+        value = row[column_to_send]
+        if value != exclude_value:
+            kafka_producer.send(topic=kafka_topic, value=str(value))
+            time.sleep(sleep_time)
 
 
 if __name__ == "__main__":