# Shelf Scribe: AI Retail Restocking & Pricing Assistant
import os
import sys
import json
import gradio as gr
# Ensure workspace paths are resolved correctly for package imports
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# Import centralized configuration parameters
from config import STATIC_DIR, IS_SPACES, MOCK_MODE
# Import decoupled subsystems
import database
import models
import reconciliation
import pdf_generator
# Initialize database schema and baseline products
database.initialize_database()
# Load CSS and JS static assets from static/ directory
CSS_PATH = os.path.join(STATIC_DIR, "style.css")
JS_PATH = os.path.join(STATIC_DIR, "script.js")
with open(CSS_PATH, "r", encoding="utf-8") as f:
custom_css = f.read()
with open(JS_PATH, "r", encoding="utf-8") as f:
custom_js = f.read()
# --- Gradio Callback Functions ---
def load_scenario_callback(scenario_name):
"""Loads demo images and config parameters based on scenario selection."""
base_dir = os.path.dirname(os.path.abspath(__file__))
if scenario_name == "Snacks Shelf & Invoice":
shelf_img = os.path.join(base_dir, "static", "demo_images", "snacks_shelf.png")
invoice_img = os.path.join(base_dir, "static", "demo_images", "invoice.png")
supplier_name = "Atlantic Wholesale Distributors"
status = "Loaded Snacks Shelf scenario! Click 'Run Audit & Reconciliation' below to process."
elif scenario_name == "Soft Drinks Shelf & Invoice":
shelf_img = os.path.join(base_dir, "static", "demo_images", "drinks_shelf.png")
invoice_img = os.path.join(base_dir, "static", "demo_images", "invoice.png")
supplier_name = "Midwest Beverages Wholesalers"
status = "Loaded Soft Drinks scenario! Click 'Run Audit & Reconciliation' below to process."
elif scenario_name == "Toiletries Shelf & Invoice":
shelf_img = os.path.join(
base_dir, "static", "demo_images", "toiletries_shelf.png"
)
invoice_img = os.path.join(base_dir, "static", "demo_images", "invoice.png")
supplier_name = "Apex Pharmacy & Beauty Distributors"
status = "Loaded Toiletries scenario! Click 'Run Audit & Reconciliation' below to process."
else:
shelf_img = None
invoice_img = None
supplier_name = "Wholesale Supplier"
status = "Scenario empty."
return shelf_img, invoice_img, supplier_name, status
def run_audit_callback(shelf_img, invoice_img, mode, scenario_name, supplier_name):
"""
Callback that runs the audit workflow either by loading cached JSON
extraction results or performing live VLM inference.
"""
shelf_detections = []
invoice_lines = []
invoice_meta = {}
base_dir = os.path.dirname(os.path.abspath(__file__))
category_map = {
"Snacks Shelf & Invoice": "snacks",
"Soft Drinks Shelf & Invoice": "drinks",
"Toiletries Shelf & Invoice": "toiletries",
}
category = category_map.get(scenario_name)
if mode == "Cached Demo Mode (Fast & Reliable)":
# Load cached JSON
cached_file_map = {
"Snacks Shelf & Invoice": "snacks_cached.json",
"Soft Drinks Shelf & Invoice": "drinks_cached.json",
"Toiletries Shelf & Invoice": "toiletries_cached.json",
}
json_file = cached_file_map.get(scenario_name, "snacks_cached.json")
json_path = os.path.join(base_dir, "demo_data", json_file)
try:
with open(json_path, "r", encoding="utf-8") as f:
cached_data = json.load(f)
shelf_detections = cached_data.get("shelf_detections", [])
invoice_lines = cached_data.get("invoice_lines", [])
invoice_meta = cached_data.get("invoice_meta", {})
if not supplier_name or supplier_name == "Wholesale Supplier":
supplier_name = invoice_meta.get("supplier_name", supplier_name)
except Exception as e:
return (
f"Failed to load cached JSON: {str(e)}
Audit error.
",
None,
None,
{"error": f"Failed to load cached data: {e}"},
)
else:
# Live Inference Mode
if not shelf_img and not invoice_img:
return (
"Please upload at least one image to run live VLM inference.
Waiting for VLM input.
",
None,
None,
{"error": "No images provided."},
)
try:
# 1. Process shelf image if provided
if shelf_img is not None:
shelf_data = models.run_visual_extraction(shelf_img, "shelf")
shelf_detections = shelf_data.get("shelf_detections", [])
# 2. Process invoice image if provided
if invoice_img is not None:
invoice_data = models.run_visual_extraction(invoice_img, "invoice")
invoice_lines = invoice_data.get("invoice_lines", [])
invoice_meta = invoice_data.get("invoice_meta", {})
if not supplier_name or supplier_name == "Wholesale Supplier":
supplier_name = invoice_meta.get("supplier_name", supplier_name)
except Exception as e:
# Live model run failed (e.g. HF_TOKEN issues, ZeroGPU timeouts)
return (
f"VLM Inference Error: {str(e)}
Please make sure your Hugging Face write token is configured correctly, or use Cached Demo Mode.
Error.
",
None,
None,
{"error": str(e)},
)
# Run deterministic reconciliation engine
reconciled = reconciliation.reconcile_operations(
shelf_detections, invoice_lines, category=category
)
# Process structured outputs for dataframes
restocks_df = [
[
r.get("product_id"),
r.get("product_name"),
r.get("action_type"),
r.get("reason"),
r.get("suggested_quantity"),
f"{int(r.get('confidence', 1.0) * 100)}%",
r.get("evidence"),
]
for r in reconciled["restocks"]
]
margin_df = [
[
m.get("product_id"),
m.get("product_name"),
m.get("action_type"),
m.get("reason"),
f"{int(m.get('confidence', 1.0) * 100)}%",
m.get("evidence"),
]
for m in reconciled["margin_warnings"]
]
label_df = [
[
lbl.get("product_id"),
lbl.get("product_name"),
lbl.get("action_type"),
lbl.get("reason"),
f"{int(lbl.get('confidence', 1.0) * 100)}%",
lbl.get("evidence"),
]
for lbl in reconciled["label_fixes"]
]
catalog_df = [
[
c.get("product_id"),
c.get("product_name"),
c.get("action_type"),
c.get("reason"),
f"{int(c.get('confidence', 1.0) * 100)}%",
c.get("evidence"),
]
for c in reconciled["catalog_warnings"]
]
# Create KPI indicators HTML
kpi_html = f"""
{len(reconciled["restocks"])}
Reorders Needed
{len(reconciled["label_fixes"])}
Label Fixes
{len(reconciled["margin_warnings"])}
Pricing Alerts
{len(reconciled["catalog_warnings"])}
Catalog Syncs
"""
# Generate supplier messaging
supplier_msg = reconciliation.generate_supplier_message(
reconciled["restocks"], supplier_name=supplier_name
)
# Also fetch all products in the loaded category to let them print labels for all shelf items!
if category:
label_products = database.db_query(
"SELECT * FROM products WHERE category = ?", (category,)
)
else:
label_products = database.db_query("SELECT * FROM products LIMIT 15")
labels_html = reconciliation.generate_printable_labels_html(label_products)
# Compile PDF downloads
po_pdf_path = pdf_generator.generate_purchase_order_pdf(
reconciled["restocks"], supplier_name=supplier_name
)
labels_pdf_path = pdf_generator.generate_shelf_labels_pdf(label_products)
raw_logs = {
"shelf_detections": shelf_detections,
"invoice_lines": invoice_lines,
"invoice_meta": invoice_meta,
"action_items": reconciled,
}
return (
kpi_html,
restocks_df,
margin_df,
label_df,
catalog_df,
supplier_msg,
labels_html,
po_pdf_path,
labels_pdf_path,
raw_logs,
)
def load_live_catalog():
"""Fetches the live catalog from SQLite to show inside editable Dataframe."""
rows = database.db_query(
"SELECT id, name, category, pack_size, supplier_price, shelf_price, target_margin_percent, reorder_threshold, notes FROM products"
)
return [
[
r["id"],
r["name"],
r["category"],
r["pack_size"],
r["supplier_price"],
r["shelf_price"],
r["target_margin_percent"],
r["reorder_threshold"],
r["notes"],
]
for r in rows
]
def save_catalog_changes(grid_data):
"""Saves edits from Gradio Dataframe grid back into SQLite products database."""
if not grid_data:
return "No catalog data found to save."
try:
for row in grid_data:
# row: [id, name, category, pack_size, supplier_price, shelf_price, target_margin_percent, reorder_threshold, notes]
db_id = row[0]
name = row[1]
category = row[2]
pack_size = int(row[3]) if row[3] is not None else 1
supplier_price = float(row[4]) if row[4] is not None else 0.0
shelf_price = float(row[5]) if row[5] is not None else 0.0
target_margin = float(row[6]) if row[6] is not None else 30.0
reorder_threshold = int(row[7]) if row[7] is not None else 5
notes = row[8]
database.db_execute(
"""
UPDATE products
SET name=?, category=?, pack_size=?, supplier_price=?,
shelf_price=?, target_margin_percent=?, reorder_threshold=?, notes=?
WHERE id=?
""",
(
name,
category,
pack_size,
supplier_price,
shelf_price,
target_margin,
reorder_threshold,
notes,
db_id,
),
)
return "Catalog changes successfully saved to database!"
except Exception as e:
return f"Error saving catalog changes: {str(e)}"
# --- Gradio UI Layout Building ---
with gr.Blocks(
title="Shelf Scribe: AI Retail Restocking Assistant", css=custom_css
) as demo:
# Header container
gr.HTML(f"""
""")
# Instruction accordion
with gr.Accordion("📖 Quick Start & Testing Instructions", open=False):
gr.Markdown("""
### How to test Shelf Scribe:
#### 1. Select a Demo Scenario
* Choose from **Snacks**, **Soft Drinks**, or **Toiletries** in the selector dropdown.
* Click **Load Scenario** to automatically fill in synthetic photos of store shelves and wholesale invoices, along with supplier details.
#### 2. Choose Processing Mode
* **Cached Demo Mode (Fast & Reliable)**: Processes the pre-computed VLM visual extractions instantly. **Highly recommended for quick evaluation!**
* **Live VLM Inference (Requires HF Token)**: Uses the warm `google/gemma-4-12B-it` model to analyze the uploaded shelf and invoice images in real-time.
#### 3. Run Audit & Analyze Results
* Click **Run Audit & Reconciliation** to process the data.
* Under **Today's Actions Dashboard**, review:
* **Restock Checklist**: Product list below threshold, with pack-size rounding.
* **Pricing & Margin Audit**: Highlights wholesale cost hikes and calculates margin degradation warnings.
* **Label Verification**: Identifies shelf price tags that mismatch the catalog or are missing entirely.
* **Supplier Message**: Instantly compiles a copy-to-clipboard WhatsApp order text.
* **Printable Labels**: Beautiful grid labels with product details, margins, and barcode layouts ready for browser printing or PDF download.
#### 4. Edit Catalog
* Change pricing thresholds or target margins in the **Catalog Editor** tab, click **Save Catalog Changes**, and re-run the audit to see recommendations recalculate dynamically!
""")
with gr.Tabs():
# TAB 1: CONTROL CENTER
with gr.Tab("1. Ingestion Control Center"):
with gr.Row():
with gr.Column(scale=1):
gr.HTML(
"1. Choose Demo Scenario
"
)
scenario_selector = gr.Dropdown(
choices=[
"Snacks Shelf & Invoice",
"Soft Drinks Shelf & Invoice",
"Toiletries Shelf & Invoice",
],
value="Snacks Shelf & Invoice",
label="Select Scenario",
)
load_scenario_btn = gr.Button(
"Load Scenario Assets", variant="secondary"
)
gr.HTML(
"2. Set Parameters
"
)
processing_mode = gr.Radio(
choices=[
"Cached Demo Mode (Fast & Reliable)",
"Live VLM Inference (Requires HF Token)",
],
value="Cached Demo Mode (Fast & Reliable)",
label="Execution Mode",
)
supplier_input = gr.Textbox(
label="Supplier Name / Contact",
placeholder="e.g. Atlantic Wholesale Distributors",
)
run_audit_btn = gr.Button(
"Run Audit & Reconciliation ⚡", variant="primary"
)
action_status = gr.Textbox(
label="Workflow Status",
value="Select a scenario and click Load.",
interactive=False,
)
with gr.Column(scale=2):
gr.HTML(
"3. Visual Inputs (Shelf & Receipt Photos)
"
)
with gr.Row():
shelf_image_input = gr.Image(
label="Supermarket Shelf Photo", type="filepath"
)
invoice_image_input = gr.Image(
label="Supplier Invoice / Receipt Photo", type="filepath"
)
# Scenario loading bindings
load_scenario_btn.click(
fn=load_scenario_callback,
inputs=[scenario_selector],
outputs=[
shelf_image_input,
invoice_image_input,
supplier_input,
action_status,
],
)
# TAB 2: ACTIONS DASHBOARD
with gr.Tab("2. Today's Actions Dashboard"):
# KPI scorecards injected here
kpi_overview_html = gr.HTML(
value="No audit executed yet. Run audit from the Ingestion Control Center tab.
Labels preview will render here after audit execution.
"
)
# Workflow trigger binding
run_audit_btn.click(
fn=run_audit_callback,
inputs=[
shelf_image_input,
invoice_image_input,
processing_mode,
scenario_selector,
supplier_input,
],
outputs=[
kpi_overview_html,
restocks_table,
margin_table,
labels_table,
catalog_table,
supplier_message_text,
labels_grid_html,
download_po_pdf,
download_labels_pdf,
action_status,
],
)
# TAB 3: CATALOG EDITOR
with gr.Tab("3. Product Catalog Editor"):
gr.Markdown("### Editable Store Product Catalog Database")
gr.Markdown(
"Edit product properties directly in the spreadsheet below. When finished, click **Save Catalog Changes** to persist details in SQLite database."
)
catalog_editor_df = gr.Dataframe(
headers=[
"ID",
"Name",
"Category",
"Pack Size",
"Supplier Cost",
"Shelf Price",
"Target Margin (%)",
"Reorder Threshold",
"Notes",
],
datatype=[
"number",
"str",
"str",
"number",
"number",
"number",
"number",
"number",
"str",
],
interactive=True,
)
with gr.Row():
load_catalog_btn = gr.Button(
"Load Catalog from DB 🔄", variant="secondary"
)
save_catalog_btn = gr.Button(
"Save Catalog Changes 💾", variant="primary"
)
catalog_save_status = gr.Textbox(
label="Save Status", value="Click Load or Save.", interactive=False
)
# Catalog editor bindings
load_catalog_btn.click(
fn=load_live_catalog, inputs=[], outputs=[catalog_editor_df]
)
save_catalog_btn.click(
fn=save_catalog_changes,
inputs=[catalog_editor_df],
outputs=[catalog_save_status],
)
# TAB 4: RAW AI LOGS
with gr.Tab("4. Raw AI Extraction Logs"):
gr.Markdown(
"#### Structured JSON outputs parsed from the Multimodal Vision Language Model."
)
raw_vlm_log_viewer = gr.JSON(label="Parsed VLM JSON Blocks")
if __name__ == "__main__":
try:
# Load live catalog database on startup inside the editor component
demo.load(fn=load_live_catalog, inputs=[], outputs=[catalog_editor_df])
if IS_SPACES:
demo.launch(
server_name="0.0.0.0",
server_port=7860,
ssr_mode=False,
css=custom_css,
js=custom_js,
)
else:
demo.launch(
server_name="127.0.0.1",
server_port=7860,
ssr_mode=False,
css=custom_css,
js=custom_js,
)
except Exception as e:
print("Error launching Shelf Scribe:", e)